espresso_node/api/

light_client.rs

use std::{collections::HashMap, fmt::Display, sync::Arc, time::Duration};

use anyhow::Result;
use committable::Committable;
use disco_types::status::StatusCode;
use espresso_types::{BlockMerkleTree, Header, NsIndex, NsProof, SeqTypes};
use futures::{TryStreamExt, future::try_join, stream::StreamExt};
use hotshot_query_service::{
    Error,
    availability::{AvailabilityDataSource, LeafQueryData, PayloadQueryData, VidCommonQueryData},
    data_source::{VersionedDataSource, storage::NodeStorage},
    merklized_state::{MerklizedStateDataSource, Snapshot},
    node::BlockId,
    types::HeightIndexed,
};
use hotshot_types::simple_certificate::QuorumCertificate2;
use itertools::izip;
use jf_merkle_tree_compat::MerkleTreeScheme;
use light_client::{
    client::NAMESPACES_PARAM_TAG,
    consensus::{header::HeaderProof, leaf::LeafProof, namespace::NamespaceProof},
};
use tagged_base64::TaggedBase64;
use versions::NEW_PROTOCOL_VERSION;

/// Construct a proof that the requested leaf is finalized.
///
/// The `finalized` hint is honored when it yields a bounded proof no longer than a direct finality
/// proof, since the client can verify hint-based proofs without signature checks.
pub(crate) async fn get_leaf_proof<State>(
    state: &State,
    requested_leaf: LeafQueryData<SeqTypes>,
    finalized: Option<usize>,
    fetch_timeout: Duration,
    chain_limit: usize,
) -> Result<LeafProof, Error>
where
    State: AvailabilityDataSource<SeqTypes> + VersionedDataSource,
    for<'a> State::ReadOnly<'a>: NodeStorage<SeqTypes>,
{
    let requested = requested_leaf.height() as usize;
    let new_protocol = requested_leaf.header().version() >= NEW_PROTOCOL_VERSION;

    let mut hint = finalized.filter(|finalized| finalized.saturating_sub(requested) <= chain_limit);

    // New-protocol chains cannot terminate early, so a nearby cert2 may yield a shorter proof.
    if let Some(finalized) = hint
        && new_protocol
        && let Some(cert2_height) = earliest_cert2_height(state, requested as u64).await
        && cert2_height + 1 < finalized as u64
    {
        hint = None;
    }

    if let Some(finalized) = hint {
        get_leaf_proof_with_finalized_assumption(
            state,
            requested_leaf,
            finalized,
            fetch_timeout,
            chain_limit,
        )
        .await
    } else if new_protocol {
        get_leaf_proof_with_cert2(state, requested_leaf, fetch_timeout, chain_limit).await
    } else {
        get_leaf_proof_with_qc_chain(state, requested_leaf, fetch_timeout, chain_limit).await
    }
}

/// The earliest stored cert2 height at or after `height`. Errors are treated as no cert2: this
/// only chooses between proof strategies, and the chosen path will surface any persistent error.
async fn earliest_cert2_height<State>(state: &State, height: u64) -> Option<u64>
where
    State: VersionedDataSource,
    for<'a> State::ReadOnly<'a>: NodeStorage<SeqTypes>,
{
    let cert2 = state
        .read()
        .await
        .ok()?
        .load_earliest_cert2(height)
        .await
        .ok()??;
    Some(cert2.data.block_number)
}

pub(crate) async fn get_leaf_proof_with_qc_chain<State>(
    state: &State,
    requested_leaf: LeafQueryData<SeqTypes>,
    fetch_timeout: Duration,
    chain_limit: usize,
) -> Result<LeafProof, Error>
where
    State: AvailabilityDataSource<SeqTypes> + VersionedDataSource,
    for<'a> State::ReadOnly<'a>: NodeStorage<SeqTypes>,
{
    let requested = requested_leaf.height() as usize;
    // Grab the endpoint and the final QC chain in the same transaction, to ensure that
    // the QC chain actually corresponds to the endpoint block (and is not subject to
    // concurrent updates).
    let mut tx = state.read().await.map_err(internal)?;
    let latest_height = NodeStorage::block_height(&mut tx).await.map_err(internal)?;
    let qc_chain = tx.latest_qc_chain().await.map_err(internal)?;
    drop(tx);

    let mut leaves = state.get_leaf_range(requested + 1..latest_height).await;
    let mut proof = LeafProof::default();
    let requested_leaf_qc = requested_leaf.qc().clone();
    proof.push(requested_leaf);

    let mut remaining = chain_limit;
    while let Some(leaf) = leaves.next().await {
        let leaf = leaf
            .with_timeout(fetch_timeout)
            .await
            .ok_or_else(|| not_found("missing leaves"))?;

        remaining = remaining
            .checked_sub(1)
            .ok_or_else(|| chain_too_long(requested, chain_limit))?;

        // HotStuff commit rules cannot terminate a chain of new-protocol leaves; switch to cert2.
        if leaf.header().version() >= NEW_PROTOCOL_VERSION {
            complete_proof_with_cert2(
                state,
                &mut proof,
                leaf,
                requested_leaf_qc.clone(),
                fetch_timeout,
                remaining,
            )
            .await?;
            return Ok(proof);
        }

        if proof.push(leaf) {
            return Ok(proof);
        }
    }

    // We reached the end of the range of interest without encountering a 3-chain. Thus, if the last
    // leaf in the chain is not already assumed finalized by the client, we must prove it finalized
    // by appending two more QCs.
    let Some([committing_qc, deciding_qc]) = qc_chain else {
        return Err(not_found("missing QC 2-chain to prove finality"));
    };
    proof.add_qc_chain(Arc::new(committing_qc), Arc::new(deciding_qc));

    Ok(proof)
}

/// Build a leaf proof for the new protocol using certificate2 finality.
///
/// Certificate2 directly commits the leaf at `cert2.data.block_number`. The indirect commit rule then
/// commits all of that leaf's uncommitted ancestors. If cert2 directly commits the requested leaf,
/// the proof contains only that leaf plus cert2. Otherwise, the proof includes the chain from the
/// requested leaf through the directly committed descendant; the verifier checks parent commitments
/// to prove the requested leaf is on that finalized chain.
pub(crate) async fn get_leaf_proof_with_cert2<State>(
    state: &State,
    requested_leaf: LeafQueryData<SeqTypes>,
    fetch_timeout: Duration,
    chain_limit: usize,
) -> Result<LeafProof, Error>
where
    State: AvailabilityDataSource<SeqTypes> + VersionedDataSource,
    for<'a> State::ReadOnly<'a>: NodeStorage<SeqTypes>,
{
    let mut proof = LeafProof::default();
    let requested_leaf_qc = requested_leaf.qc().clone();
    complete_proof_with_cert2(
        state,
        &mut proof,
        requested_leaf,
        requested_leaf_qc,
        fetch_timeout,
        chain_limit,
    )
    .await?;
    Ok(proof)
}

/// Extend `proof` from `leaf` to the leaf directly committed by the earliest stored cert2, adding
/// at most `chain_limit` leaves.
async fn complete_proof_with_cert2<State>(
    state: &State,
    proof: &mut LeafProof,
    leaf: LeafQueryData<SeqTypes>,
    requested_leaf_qc: QuorumCertificate2<SeqTypes>,
    fetch_timeout: Duration,
    chain_limit: usize,
) -> Result<(), Error>
where
    State: AvailabilityDataSource<SeqTypes> + VersionedDataSource,
    for<'a> State::ReadOnly<'a>: NodeStorage<SeqTypes>,
{
    let start_height = leaf.height();
    let start_commit = leaf.leaf().commit();

    // The new leaf may already complete a HotStuff chain begun before the protocol cutover.
    if proof.push(leaf) {
        return Ok(());
    }

    let cert2 = state
        .read()
        .await
        .map_err(internal)?
        .load_earliest_cert2(start_height)
        .await
        .map_err(internal)?
        .ok_or_else(|| {
            not_found(format!(
                "no cert2 finality proof available at or after height {start_height}"
            ))
        })?;

    let cert2_height = cert2.data.block_number;
    if cert2_height < start_height {
        return Err(not_found(
            "cert2 finality proof is older than requested leaf",
        ));
    }
    if cert2_height - start_height > chain_limit as u64 {
        return Err(not_found(format!(
            "earliest cert2 finality proof (height {cert2_height}) is more than {chain_limit} \
             leaves past height {start_height}"
        )));
    }

    if cert2_height == start_height {
        if start_commit != cert2.data.leaf_commit {
            return Err(internal("stored cert2 does not finalize the expected leaf"));
        }
        proof.add_certificate(Arc::new(cert2), requested_leaf_qc);
        return Ok(());
    }

    let mut leaves = state
        .get_leaf_range(start_height as usize + 1..cert2_height as usize + 1)
        .await;
    // Extend the proof chain until we reach the leaf directly committed by cert2. Once that leaf is
    // present and matches cert2's commitment, the requested leaf is finalized by the indirect
    // commit rule.
    while let Some(leaf) = leaves.next().await {
        let leaf = leaf
            .with_timeout(fetch_timeout)
            .await
            .ok_or_else(|| not_found("missing leaves"))?;

        if leaf.height() == cert2_height {
            if leaf.leaf().commit() != cert2.data.leaf_commit {
                return Err(internal("stored cert2 does not finalize the expected leaf"));
            }
            if !proof.push(leaf) {
                proof.add_certificate(Arc::new(cert2), requested_leaf_qc);
            }
            return Ok(());
        }
        if proof.push(leaf) {
            return Ok(());
        }
    }

    Err(not_found("missing cert2 leaf"))
}

pub(crate) async fn get_leaf_proof_with_finalized_assumption<State>(
    state: &State,
    requested_leaf: LeafQueryData<SeqTypes>,
    finalized: usize,
    fetch_timeout: Duration,
    chain_limit: usize,
) -> Result<LeafProof, Error>
where
    State: AvailabilityDataSource<SeqTypes>,
{
    let requested = requested_leaf.height() as usize;
    // If we have a known-finalized block, we will not need a final 2-chain of QCs to prove
    // the last leaf in the result finalized, since we will either terminate with a 3-chain
    // of leaves or at the `finalized` leaf.
    if finalized <= requested {
        return Err(Error::Custom {
            message: format!(
                "finalized leaf height ({finalized}) must be greater than requested ({requested})"
            ),
            status: StatusCode::BAD_REQUEST,
        });
    }
    if finalized - requested > chain_limit {
        return Err(Error::Custom {
            message: format!(
                "finalized leaf height ({finalized}) is more than {chain_limit} blocks past the \
                 requested leaf ({requested}); request a proof without the finalized parameter \
                 instead"
            ),
            status: StatusCode::BAD_REQUEST,
        });
    }

    let mut leaves = state.get_leaf_range(requested + 1..finalized).await;
    let mut proof = LeafProof::default();
    proof.push(requested_leaf);

    while let Some(leaf) = leaves.next().await {
        let leaf = leaf
            .with_timeout(fetch_timeout)
            .await
            .ok_or_else(|| not_found("missing leaves"))?;

        if proof.push(leaf) {
            return Ok(proof);
        }
    }

    Ok(proof)
}

pub(crate) async fn get_header_proof<State>(
    state: &State,
    root: u64,
    requested: BlockId<SeqTypes>,
    fetch_timeout: Duration,
) -> Result<HeaderProof, Error>
where
    State: AvailabilityDataSource<SeqTypes>
        + MerklizedStateDataSource<SeqTypes, BlockMerkleTree, { BlockMerkleTree::ARITY }>
        + VersionedDataSource,
{
    let header = state
        .get_header(requested)
        .await
        .with_timeout(fetch_timeout)
        .await
        .ok_or_else(|| not_found(format!("unknown header {requested}")))?;
    if header.height() >= root {
        return Err(Error::Custom {
            message: format!(
                "height ({}) must be less than root ({root})",
                header.height()
            ),
            status: StatusCode::BAD_REQUEST,
        });
    }
    let path = MerklizedStateDataSource::<SeqTypes, BlockMerkleTree, _>::get_path(
        state,
        Snapshot::Index(root),
        header.height(),
    )
    .await
    .map_err(|source| Error::MerklizedState {
        source: source.into(),
    })?;

    Ok(HeaderProof::new(header, path))
}

pub(crate) async fn fetch_block_data_range<State>(
    state: &State,
    start: usize,
    end: usize,
    fetch_timeout: Duration,
    large_object_range_limit: usize,
) -> Result<
    Vec<(
        Header,
        PayloadQueryData<SeqTypes>,
        VidCommonQueryData<SeqTypes>,
    )>,
    Error,
>
where
    State: AvailabilityDataSource<SeqTypes>,
{
    if end <= start {
        return Err(Error::Custom {
            message: format!("requested empty interval [{start}, {end})"),
            status: StatusCode::BAD_REQUEST,
        });
    }
    if end - start > large_object_range_limit {
        return Err(Error::Custom {
            message: format!(
                "requested range [{start}, {end}) exceeds maximum size {large_object_range_limit}"
            ),
            status: StatusCode::BAD_REQUEST,
        });
    }

    let fetch_headers = async move {
        state
            .get_header_range(start..end)
            .await
            .enumerate()
            .then(|(i, fetch)| async move {
                fetch
                    .with_timeout(fetch_timeout)
                    .await
                    .ok_or_else(|| Error::Custom {
                        message: format!("missing header {}", start + i),
                        status: StatusCode::NOT_FOUND,
                    })
            })
            .try_collect::<Vec<_>>()
            .await
    };
    let fetch_payloads = async move {
        state
            .get_payload_range(start..end)
            .await
            .enumerate()
            .then(|(i, fetch)| async move {
                fetch
                    .with_timeout(fetch_timeout)
                    .await
                    .ok_or_else(|| Error::Custom {
                        message: format!("missing payload {}", start + i),
                        status: StatusCode::NOT_FOUND,
                    })
            })
            .try_collect::<Vec<_>>()
            .await
    };
    let fetch_vid_commons = async move {
        state
            .get_vid_common_range(start..end)
            .await
            .enumerate()
            .then(|(i, fetch)| async move {
                fetch
                    .with_timeout(fetch_timeout)
                    .await
                    .ok_or_else(|| Error::Custom {
                        message: format!("missing VID common {}", start + i),
                        status: StatusCode::NOT_FOUND,
                    })
            })
            .try_collect::<Vec<_>>()
            .await
    };
    let (headers, (payloads, vid_commons)) =
        try_join(fetch_headers, try_join(fetch_payloads, fetch_vid_commons)).await?;

    Ok(izip!(headers, payloads, vid_commons).collect())
}

/// Single-namespace version of [`get_namespaces_proof_range`].
pub(crate) async fn get_namespace_proof_range<State>(
    state: &State,
    start: usize,
    end: usize,
    namespace: u64,
    fetch_timeout: Duration,
    large_object_range_limit: usize,
) -> Result<Vec<NamespaceProof>, Error>
where
    State: AvailabilityDataSource<SeqTypes>,
{
    let blocks = get_namespaces_proof_range(
        state,
        start,
        end,
        &[namespace],
        fetch_timeout,
        large_object_range_limit,
    )
    .await?;
    Ok(blocks
        .into_iter()
        .map(|mut block| {
            block
                .remove(&namespace)
                .unwrap_or_else(NamespaceProof::not_present)
        })
        .collect())
}

pub(crate) async fn get_namespaces_proof_range<State>(
    state: &State,
    start: usize,
    end: usize,
    namespaces: &[u64],
    fetch_timeout: Duration,
    large_object_range_limit: usize,
) -> Result<Vec<HashMap<u64, NamespaceProof>>, Error>
where
    State: AvailabilityDataSource<SeqTypes>,
{
    fetch_block_data_range(state, start, end, fetch_timeout, large_object_range_limit)
        .await?
        .into_iter()
        .map(|(header, payload, vid_common)| {
            namespaces
                .iter()
                .filter_map(|&namespace| {
                    let ns_index = header.ns_table().find_ns_id(&namespace.into())?;
                    Some(
                        build_namespace_proof(&payload, &ns_index, &vid_common)
                            .map(|proof| (namespace, proof)),
                    )
                })
                .collect::<Result<HashMap<_, _>, _>>()
        })
        .collect()
}

/// Decode the `namespaces` path segment: a `TaggedBase64` string tagged `NS` wrapping a
/// JSON-encoded `Vec<u64>`.
pub(crate) fn parse_namespaces_str(encoded: &str) -> anyhow::Result<Vec<u64>> {
    let encoded: TaggedBase64 = encoded
        .parse()
        .map_err(|err| anyhow::anyhow!("invalid namespaces parameter: {err}"))?;
    if encoded.tag() != NAMESPACES_PARAM_TAG {
        anyhow::bail!(
            "invalid namespaces parameter tag: expected {NAMESPACES_PARAM_TAG}, got {}",
            encoded.tag()
        );
    }
    serde_json::from_slice(&encoded.value())
        .map_err(|err| anyhow::anyhow!("invalid namespaces parameter: {err}"))
}

/// Construct a [`NamespaceProof`] for the namespace at `ns_index` of the given block.
fn build_namespace_proof(
    payload: &PayloadQueryData<SeqTypes>,
    ns_index: &NsIndex,
    vid_common: &VidCommonQueryData<SeqTypes>,
) -> Result<NamespaceProof, Error> {
    let ns_proof =
        NsProof::new(payload.data(), ns_index, vid_common.common()).ok_or_else(|| {
            Error::Custom {
                message: "failed to construct namespace proof".into(),
                status: StatusCode::INTERNAL_SERVER_ERROR,
            }
        })?;
    Ok(NamespaceProof::new(ns_proof, vid_common.common().clone()))
}

fn internal(err: impl Display) -> Error {
    Error::Custom {
        message: err.to_string(),
        status: StatusCode::INTERNAL_SERVER_ERROR,
    }
}

fn not_found(msg: impl Into<String>) -> Error {
    Error::Custom {
        message: msg.into(),
        status: StatusCode::NOT_FOUND,
    }
}

fn chain_too_long(requested: usize, chain_limit: usize) -> Error {
    not_found(format!(
        "no finality proof found within {chain_limit} leaves of requested leaf {requested}"
    ))
}

#[cfg(test)]
mod test {
    use std::marker::PhantomData;

    use committable::Committable;
    use disco_types::error::Error;
    use espresso_types::BLOCK_MERKLE_TREE_HEIGHT;
    use futures::future::join_all;
    use hotshot_query_service::{
        availability::{BlockQueryData, TransactionIndex, VidCommonQueryData},
        data_source::{Transaction, storage::UpdateAvailabilityStorage},
        merklized_state::UpdateStateData,
    };
    use hotshot_types::{
        data::ViewNumber, simple_certificate::CertificatePair, simple_vote::Vote2Data,
    };
    use jf_merkle_tree_compat::{AppendableMerkleTreeScheme, ToTraversalPath};
    use light_client::{
        consensus::leaf::{FinalityProof, LeafProofHint},
        testing::{
            AlwaysTrueQuorum, ENABLE_EPOCHS, LEGACY_VERSION, TestClient, VersionCheckQuorum,
            custom_leaf_chain_with_upgrade, leaf_chain, leaf_chain_with_upgrade,
        },
    };
    use versions::{DRB_AND_HEADER_UPGRADE_VERSION, EPOCH_VERSION, NEW_PROTOCOL_VERSION, Upgrade};

    use super::*;
    use crate::api::{
        data_source::{SequencerDataSource, testing::TestableSequencerDataSource},
        sql::DataSource,
    };

    const CHAIN_LIMIT: usize = 500;

    fn cert2_for_leaf(leaf: &LeafQueryData<SeqTypes>) -> espresso_types::Certificate2<SeqTypes> {
        let data = Vote2Data {
            leaf_commit: leaf.leaf().commit(),
            epoch: leaf.qc().data.epoch.unwrap(),
            block_number: leaf.height(),
        };
        espresso_types::Certificate2::new(
            data.clone(),
            data.commit(),
            leaf.leaf().view_number(),
            None,
            PhantomData,
        )
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_two_chain() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // Insert some leaves, forming a chain.
        let leaves = leaf_chain(1..=3, EPOCH_VERSION).await;
        {
            let mut tx = ds.write().await.unwrap();
            tx.insert_leaf(&leaves[0]).await.unwrap();
            tx.insert_leaf(&leaves[1]).await.unwrap();
            tx.insert_leaf(&leaves[2]).await.unwrap();
            tx.commit().await.unwrap();
        }

        // Ask for the first leaf; it is proved finalized by the chain formed along with the second.
        let proof =
            get_leaf_proof_with_qc_chain(&ds, leaves[0].clone(), Duration::MAX, CHAIN_LIMIT)
                .await
                .unwrap();
        assert_eq!(
            proof
                .verify(LeafProofHint::Quorum(&AlwaysTrueQuorum))
                .await
                .unwrap(),
            leaves[0]
        );
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_finalized() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // Insert a single leaf. We will not be able to provide proofs ending in a leaf chain, but
        // we can return a leaf if the leaf after it is already known to be finalized.
        let leaves = leaf_chain(1..=2, EPOCH_VERSION).await;
        {
            let mut tx = ds.write().await.unwrap();
            tx.insert_leaf(&leaves[0]).await.unwrap();
            tx.commit().await.unwrap();
        }

        let proof = get_leaf_proof_with_finalized_assumption(
            &ds,
            leaves[0].clone(),
            2,
            Duration::MAX,
            CHAIN_LIMIT,
        )
        .await
        .unwrap();
        assert_eq!(
            proof
                .verify(LeafProofHint::assumption(leaves[1].leaf()))
                .await
                .unwrap(),
            leaves[0]
        );
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_new_protocol_finalized_assumption() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        let leaves = leaf_chain(1..=2, NEW_PROTOCOL_VERSION).await;
        {
            let mut tx = ds.write().await.unwrap();
            tx.insert_leaf(&leaves[0]).await.unwrap();
            tx.commit().await.unwrap();
        }

        let proof = get_leaf_proof_with_finalized_assumption(
            &ds,
            leaves[0].clone(),
            2,
            Duration::MAX,
            CHAIN_LIMIT,
        )
        .await
        .unwrap();
        assert!(matches!(proof.proof(), FinalityProof::Assumption));
        assert_eq!(
            proof
                .verify(LeafProofHint::assumption(leaves[1].leaf()))
                .await
                .unwrap(),
            leaves[0]
        );
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_new_protocol_cert2() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        let leaves = leaf_chain(1..=2, NEW_PROTOCOL_VERSION).await;
        let cert2_leaf = &leaves[1];
        let cert2 = cert2_for_leaf(cert2_leaf);

        {
            let mut tx = ds.write().await.unwrap();
            tx.insert_leaf(&leaves[0]).await.unwrap();
            tx.insert_leaf(cert2_leaf).await.unwrap();
            tx.insert_cert2(cert2_leaf.height(), cert2).await.unwrap();
            tx.commit().await.unwrap();
        }

        let proof = get_leaf_proof_with_cert2(&ds, leaves[0].clone(), Duration::MAX, CHAIN_LIMIT)
            .await
            .unwrap();
        assert!(matches!(proof.proof(), FinalityProof::NewProtocol { .. }));
        assert_eq!(
            proof
                .verify(LeafProofHint::Quorum(&AlwaysTrueQuorum))
                .await
                .unwrap(),
            leaves[0]
        );
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_new_protocol_cert2_chain_limit() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // A proof for the first leaf must extend to the nearest cert2, three leaves away.
        let leaves = leaf_chain(1..=4, NEW_PROTOCOL_VERSION).await;
        let cert2_leaf = &leaves[3];
        let cert2 = cert2_for_leaf(cert2_leaf);

        {
            let mut tx = ds.write().await.unwrap();
            for leaf in &leaves {
                tx.insert_leaf(leaf).await.unwrap();
            }
            tx.insert_cert2(cert2_leaf.height(), cert2).await.unwrap();
            tx.commit().await.unwrap();
        }

        let err = get_leaf_proof_with_cert2(&ds, leaves[0].clone(), Duration::MAX, 2)
            .await
            .unwrap_err();
        assert_eq!(err.status(), StatusCode::NOT_FOUND);

        let proof = get_leaf_proof_with_cert2(&ds, leaves[0].clone(), Duration::MAX, 3)
            .await
            .unwrap();
        assert!(matches!(proof.proof(), FinalityProof::NewProtocol { .. }));
        assert_eq!(
            proof
                .verify(LeafProofHint::Quorum(&AlwaysTrueQuorum))
                .await
                .unwrap(),
            leaves[0]
        );
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_new_protocol_hint_vs_cert2() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        let leaves = leaf_chain(1..=5, NEW_PROTOCOL_VERSION).await;
        let cert2_leaf = &leaves[1];
        let cert2 = cert2_for_leaf(cert2_leaf);

        {
            let mut tx = ds.write().await.unwrap();
            for leaf in &leaves {
                tx.insert_leaf(leaf).await.unwrap();
            }
            tx.insert_cert2(cert2_leaf.height(), cert2).await.unwrap();
            tx.commit().await.unwrap();
        }

        // The cert2 at height 2 beats the hint at height 5, so the hint is ignored.
        let proof = get_leaf_proof(&ds, leaves[0].clone(), Some(5), Duration::MAX, CHAIN_LIMIT)
            .await
            .unwrap();
        assert!(matches!(proof.proof(), FinalityProof::NewProtocol { .. }));
        assert_eq!(
            proof
                .verify(LeafProofHint::Quorum(&AlwaysTrueQuorum))
                .await
                .unwrap(),
            leaves[0]
        );

        // A hint at height 3 ties the cert2 proof length, so the hint wins.
        let proof = get_leaf_proof(&ds, leaves[0].clone(), Some(3), Duration::MAX, CHAIN_LIMIT)
            .await
            .unwrap();
        assert!(matches!(proof.proof(), FinalityProof::Assumption));
        assert_eq!(
            proof
                .verify(LeafProofHint::assumption(leaves[2].leaf()))
                .await
                .unwrap(),
            leaves[0]
        );

        // With no cert2 at or after the requested height, the hint is honored.
        let proof = get_leaf_proof(&ds, leaves[2].clone(), Some(5), Duration::MAX, CHAIN_LIMIT)
            .await
            .unwrap();
        assert!(matches!(proof.proof(), FinalityProof::Assumption));
        assert_eq!(
            proof
                .verify(LeafProofHint::assumption(leaves[4].leaf()))
                .await
                .unwrap(),
            leaves[2]
        );
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_distant_hint_falls_through() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // Legacy leaves 1-3 upgrade to the new protocol at height 4, with a cert2 at height 5, so
        // both fall-through paths have a bounded proof available.
        let leaves = leaf_chain_with_upgrade(
            1..=5,
            4,
            Upgrade::new(DRB_AND_HEADER_UPGRADE_VERSION, NEW_PROTOCOL_VERSION),
        )
        .await;
        let cert2_leaf = &leaves[4];
        let cert2 = cert2_for_leaf(cert2_leaf);

        {
            let mut tx = ds.write().await.unwrap();
            for leaf in &leaves {
                tx.insert_leaf(leaf).await.unwrap();
            }
            tx.insert_cert2(cert2_leaf.height(), cert2).await.unwrap();
            tx.commit().await.unwrap();
        }

        // A hint more than `chain_limit` past a legacy leaf is ignored in favor of a QC chain.
        let proof = get_leaf_proof(
            &ds,
            leaves[0].clone(),
            Some(1000),
            Duration::MAX,
            CHAIN_LIMIT,
        )
        .await
        .unwrap();
        assert!(matches!(proof.proof(), FinalityProof::HotStuff2 { .. }));
        assert_eq!(
            proof
                .verify(LeafProofHint::Quorum(&AlwaysTrueQuorum))
                .await
                .unwrap(),
            leaves[0]
        );

        // A hint more than `chain_limit` past a new-protocol leaf is ignored in favor of cert2.
        let proof = get_leaf_proof(
            &ds,
            leaves[3].clone(),
            Some(1000),
            Duration::MAX,
            CHAIN_LIMIT,
        )
        .await
        .unwrap();
        assert!(matches!(proof.proof(), FinalityProof::NewProtocol { .. }));
        assert_eq!(
            proof
                .verify(LeafProofHint::Quorum(&AlwaysTrueQuorum))
                .await
                .unwrap(),
            leaves[3]
        );
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_qc_chain_chain_limit() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // Proving the first leaf finalized requires walking the two subsequent leaves.
        let leaves = leaf_chain(1..=3, EPOCH_VERSION).await;
        {
            let mut tx = ds.write().await.unwrap();
            for leaf in &leaves {
                tx.insert_leaf(leaf).await.unwrap();
            }
            tx.commit().await.unwrap();
        }

        let err = get_leaf_proof_with_qc_chain(&ds, leaves[0].clone(), Duration::MAX, 1)
            .await
            .unwrap_err();
        assert_eq!(err.status(), StatusCode::NOT_FOUND);

        let proof = get_leaf_proof_with_qc_chain(&ds, leaves[0].clone(), Duration::MAX, 2)
            .await
            .unwrap();
        assert_eq!(
            proof
                .verify(LeafProofHint::Quorum(&AlwaysTrueQuorum))
                .await
                .unwrap(),
            leaves[0]
        );
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_finalized_hint_too_far() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        let leaves = leaf_chain(1..=2, EPOCH_VERSION).await;
        {
            let mut tx = ds.write().await.unwrap();
            tx.insert_leaf(&leaves[0]).await.unwrap();
            tx.commit().await.unwrap();
        }

        let err = get_leaf_proof_with_finalized_assumption(
            &ds,
            leaves[0].clone(),
            1000,
            Duration::MAX,
            10,
        )
        .await
        .unwrap_err();
        assert_eq!(err.status(), StatusCode::BAD_REQUEST);
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_qc_chain_new_protocol_cutover() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // Upgrade mid-chain and skew view numbers so no HotStuff 2-chain ever forms; the proof
        // for the first (pre-upgrade) leaf must fall through to cert2 finality.
        let leaves = custom_leaf_chain_with_upgrade(
            1..=4,
            2,
            Upgrade::new(DRB_AND_HEADER_UPGRADE_VERSION, NEW_PROTOCOL_VERSION),
            |proposal| {
                proposal.view_number = ViewNumber::new(proposal.block_header.height() * 2);
            },
        )
        .await;
        assert_eq!(leaves[0].header().version(), DRB_AND_HEADER_UPGRADE_VERSION);
        assert_eq!(leaves[1].header().version(), NEW_PROTOCOL_VERSION);
        let cert2_leaf = &leaves[3];
        let cert2 = cert2_for_leaf(cert2_leaf);

        {
            let mut tx = ds.write().await.unwrap();
            for leaf in &leaves {
                tx.insert_leaf(leaf).await.unwrap();
            }
            tx.insert_cert2(cert2_leaf.height(), cert2).await.unwrap();
            tx.commit().await.unwrap();
        }

        let proof =
            get_leaf_proof_with_qc_chain(&ds, leaves[0].clone(), Duration::MAX, CHAIN_LIMIT)
                .await
                .unwrap();
        assert!(matches!(proof.proof(), FinalityProof::NewProtocol { .. }));
        assert_eq!(
            proof
                .verify(LeafProofHint::Quorum(&AlwaysTrueQuorum))
                .await
                .unwrap(),
            leaves[0]
        );
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_qc_chain_new_protocol_cutover_two_chain() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // Upgrade at height 3: the HotStuff2 chain begun before the cutover completes without
        // needing a cert2 (none is stored).
        let leaves = leaf_chain_with_upgrade(
            1..=3,
            3,
            Upgrade::new(DRB_AND_HEADER_UPGRADE_VERSION, NEW_PROTOCOL_VERSION),
        )
        .await;
        assert_eq!(leaves[1].header().version(), DRB_AND_HEADER_UPGRADE_VERSION);
        assert_eq!(leaves[2].header().version(), NEW_PROTOCOL_VERSION);

        {
            let mut tx = ds.write().await.unwrap();
            for leaf in &leaves {
                tx.insert_leaf(leaf).await.unwrap();
            }
            tx.commit().await.unwrap();
        }

        let proof =
            get_leaf_proof_with_qc_chain(&ds, leaves[0].clone(), Duration::MAX, CHAIN_LIMIT)
                .await
                .unwrap();
        assert!(matches!(proof.proof(), FinalityProof::HotStuff2 { .. }));
        assert_eq!(
            proof
                .verify(LeafProofHint::Quorum(&VersionCheckQuorum::new(
                    leaves.iter().map(|leaf| leaf.leaf().clone())
                )))
                .await
                .unwrap(),
            leaves[0]
        );
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_qc_chain_last_legacy_leaf_uses_cert2() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // Upgrade at height 2: any 2-chain proving the last legacy leaf spans the cutover, so the
        // prover must fall through to the cert2 at height 3 instead.
        let leaves = leaf_chain_with_upgrade(
            1..=4,
            2,
            Upgrade::new(DRB_AND_HEADER_UPGRADE_VERSION, NEW_PROTOCOL_VERSION),
        )
        .await;
        assert_eq!(leaves[0].header().version(), DRB_AND_HEADER_UPGRADE_VERSION);
        assert_eq!(leaves[1].header().version(), NEW_PROTOCOL_VERSION);

        let cert2_leaf = &leaves[2];
        let cert2 = cert2_for_leaf(cert2_leaf);

        {
            let mut tx = ds.write().await.unwrap();
            for leaf in &leaves {
                tx.insert_leaf(leaf).await.unwrap();
            }
            tx.insert_cert2(cert2_leaf.height(), cert2).await.unwrap();
            tx.commit().await.unwrap();
        }

        let proof =
            get_leaf_proof_with_qc_chain(&ds, leaves[0].clone(), Duration::MAX, CHAIN_LIMIT)
                .await
                .unwrap();
        assert!(matches!(proof.proof(), FinalityProof::NewProtocol { .. }));
        assert_eq!(
            proof
                .verify(LeafProofHint::Quorum(&VersionCheckQuorum::new(
                    leaves.iter().map(|leaf| leaf.leaf().clone())
                )))
                .await
                .unwrap(),
            leaves[0]
        );
    }

    /// With a gapless cutover, the leaf two before the cutover (`upgrade_height
    /// - 2`) is finalized by the old protocol: its HotStuff2 2-chain completes
    /// within pre-cutover leaves (deciding QC belongs to `upgrade_height - 1`,
    /// still V5), so no cert2 is needed.
    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_leaf_before_cutover_uses_hotstuff2() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // Cutover at height 4, so `upgrade_height - 2` is height 2 (leaves[1]).
        let leaves = leaf_chain_with_upgrade(
            1..=5,
            4,
            Upgrade::new(DRB_AND_HEADER_UPGRADE_VERSION, NEW_PROTOCOL_VERSION),
        )
        .await;
        assert!(leaves[1].header().version() < NEW_PROTOCOL_VERSION);
        assert!(leaves[3].header().version() >= NEW_PROTOCOL_VERSION);

        {
            let mut tx = ds.write().await.unwrap();
            for leaf in &leaves {
                tx.insert_leaf(leaf).await.unwrap();
            }
            tx.commit().await.unwrap();
        }

        let proof =
            get_leaf_proof_with_qc_chain(&ds, leaves[1].clone(), Duration::MAX, CHAIN_LIMIT)
                .await
                .unwrap();
        assert!(matches!(proof.proof(), FinalityProof::HotStuff2 { .. }));
        assert_eq!(
            proof
                .verify(LeafProofHint::Quorum(&VersionCheckQuorum::new(
                    leaves.iter().map(|leaf| leaf.leaf().clone())
                )))
                .await
                .unwrap(),
            leaves[1]
        );
    }

    /// When a view gap at the cutover prevents a HotStuff2 2-chain from
    /// completing within pre-cutover leaves, the leaf two before the cutover
    /// (`upgrade_height - 2`) instead falls through to cert2 finality. Skewing
    /// the view numbers breaks the consecutive-view requirement, so no 2-chain
    /// forms and the proof terminates at the first cert2.
    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_leaf_before_cutover_uses_cert2() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // Cutover at height 4, so `upgrade_height - 2` is height 2 (leaves[1]).
        // Skewed views mean no HotStuff2 2-chain ever forms.
        let leaves = custom_leaf_chain_with_upgrade(
            1..=5,
            4,
            Upgrade::new(DRB_AND_HEADER_UPGRADE_VERSION, NEW_PROTOCOL_VERSION),
            |proposal| {
                proposal.view_number = ViewNumber::new(proposal.block_header.height() * 2);
            },
        )
        .await;
        assert!(leaves[1].header().version() < NEW_PROTOCOL_VERSION);
        assert!(leaves[3].header().version() >= NEW_PROTOCOL_VERSION);
        let cert2_leaf = &leaves[3];
        let cert2 = cert2_for_leaf(cert2_leaf);

        {
            let mut tx = ds.write().await.unwrap();
            for leaf in &leaves {
                tx.insert_leaf(leaf).await.unwrap();
            }
            tx.insert_cert2(cert2_leaf.height(), cert2).await.unwrap();
            tx.commit().await.unwrap();
        }

        let proof =
            get_leaf_proof_with_qc_chain(&ds, leaves[1].clone(), Duration::MAX, CHAIN_LIMIT)
                .await
                .unwrap();
        assert!(matches!(proof.proof(), FinalityProof::NewProtocol { .. }));
        assert_eq!(
            proof
                .verify(LeafProofHint::Quorum(&AlwaysTrueQuorum))
                .await
                .unwrap(),
            leaves[1]
        );
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_bad_finalized() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // Insert a single leaf. If we request this leaf but provide a finalized leaf which is
        // earlier, we should fail.
        let leaves = leaf_chain(1..2, EPOCH_VERSION).await;
        {
            let mut tx = ds.write().await.unwrap();
            tx.insert_leaf(&leaves[0]).await.unwrap();
            tx.commit().await.unwrap();
        }

        let err = get_leaf_proof_with_finalized_assumption(
            &ds,
            leaves[0].clone(),
            0,
            Duration::MAX,
            CHAIN_LIMIT,
        )
        .await
        .unwrap_err();
        assert_eq!(err.status(), StatusCode::BAD_REQUEST);
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_no_chain() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // Insert multiple leaves that don't chain. We will not be able to prove these are
        // finalized.
        let leaves = leaf_chain(1..=4, EPOCH_VERSION).await;
        {
            let mut tx = ds.write().await.unwrap();
            tx.insert_leaf(&leaves[0]).await.unwrap();
            tx.insert_leaf(&leaves[2]).await.unwrap();
            tx.insert_leaf(&leaves[3]).await.unwrap();
            tx.commit().await.unwrap();
        }

        let err = get_leaf_proof_with_qc_chain(
            &ds,
            leaves[0].clone(),
            Duration::from_secs(1),
            CHAIN_LIMIT,
        )
        .await
        .unwrap_err();
        assert_eq!(err.status(), StatusCode::NOT_FOUND);

        // Even if we start from a finalized leave that extends one of the leaves we do have (4,
        // extends 3) we fail to generate a proof because we can't generate a chain from the
        // requested leaf (1) to the finalized leaf (4), since leaf 2 is missing.
        let err = get_leaf_proof_with_finalized_assumption(
            &ds,
            leaves[0].clone(),
            4,
            Duration::from_secs(1),
            CHAIN_LIMIT,
        )
        .await
        .unwrap_err();
        assert_eq!(err.status(), StatusCode::NOT_FOUND);
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_final_qcs() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // Insert a single leaf, plus an extra QC chain proving it finalized.
        let leaves = leaf_chain(1..=3, EPOCH_VERSION).await;
        let qcs = [
            CertificatePair::for_parent(leaves[1].leaf()),
            CertificatePair::for_parent(leaves[2].leaf()),
        ];
        {
            let mut tx = ds.write().await.unwrap();
            tx.insert_leaf_with_qc_chain(&leaves[0], Some(qcs.clone()))
                .await
                .unwrap();
            tx.commit().await.unwrap();
        }

        let proof =
            get_leaf_proof_with_qc_chain(&ds, leaves[0].clone(), Duration::MAX, CHAIN_LIMIT)
                .await
                .unwrap();
        assert_eq!(
            proof
                .verify(LeafProofHint::Quorum(&AlwaysTrueQuorum))
                .await
                .unwrap(),
            leaves[0]
        );
    }

    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    async fn test_upgrade_to_epochs() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // Upgrade to epochs (and enabling HotStuff2) in the middle of a leaf chain, so that the
        // last leaf in the chain only requires 2 QCs to verify, even though at the start of the
        // chain we would have required 3.
        let leaves = leaf_chain_with_upgrade(1..=4, 2, ENABLE_EPOCHS).await;
        assert_eq!(leaves[0].header().version(), LEGACY_VERSION);
        assert_eq!(leaves[1].header().version(), DRB_AND_HEADER_UPGRADE_VERSION);
        let qcs = [
            CertificatePair::for_parent(leaves[2].leaf()),
            CertificatePair::for_parent(leaves[3].leaf()),
        ];
        {
            let mut tx = ds.write().await.unwrap();
            tx.insert_leaf(&leaves[0]).await.unwrap();
            tx.insert_leaf_with_qc_chain(&leaves[1], Some(qcs.clone()))
                .await
                .unwrap();
            tx.commit().await.unwrap();
        }

        let proof =
            get_leaf_proof_with_qc_chain(&ds, leaves[0].clone(), Duration::MAX, CHAIN_LIMIT)
                .await
                .unwrap();
        assert_eq!(
            proof
                .verify(LeafProofHint::Quorum(&VersionCheckQuorum::new(
                    leaves.iter().map(|leaf| leaf.leaf().clone())
                )))
                .await
                .unwrap(),
            leaves[0]
        );
        assert!(matches!(proof.proof(), FinalityProof::HotStuff2 { .. }))
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_header_proof() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // Construct a chain of leaves, plus the corresponding block Merkle tree at each leaf.
        let leaves = leaf_chain(0..=2, EPOCH_VERSION).await;
        let mts = leaves
            .iter()
            .scan(
                BlockMerkleTree::new(BLOCK_MERKLE_TREE_HEIGHT),
                |mt, leaf| {
                    assert_eq!(mt.commitment(), leaf.header().block_merkle_tree_root());
                    let item = mt.clone();
                    mt.push(leaf.block_hash()).unwrap();
                    Some(item)
                },
            )
            .collect::<Vec<_>>();

        // Save all those objects in the DB.
        {
            let mut tx = ds.write().await.unwrap();
            for (leaf, mt) in leaves.iter().zip(&mts) {
                tx.insert_leaf(leaf).await.unwrap();

                if leaf.height() > 0 {
                    let merkle_path = mt.lookup(leaf.height() - 1).expect_ok().unwrap().1;
                    UpdateStateData::<SeqTypes, BlockMerkleTree, _>::insert_merkle_nodes(
                        &mut tx,
                        merkle_path,
                        ToTraversalPath::<{ BlockMerkleTree::ARITY }>::to_traversal_path(
                            &(leaf.height() - 1),
                            BLOCK_MERKLE_TREE_HEIGHT,
                        ),
                        leaf.height(),
                    )
                    .await
                    .unwrap();
                    UpdateStateData::<SeqTypes, BlockMerkleTree, _>::set_last_state_height(
                        &mut tx,
                        leaf.height() as usize,
                    )
                    .await
                    .unwrap();
                }
            }
            tx.commit().await.unwrap();
        }

        // Test happy path.
        for (root, mt) in mts.iter().enumerate().skip(1) {
            for (height, leaf) in leaves.iter().enumerate().take(root) {
                tracing::info!(root, height, "test happy path");
                let proof =
                    get_header_proof(&ds, root as u64, BlockId::Number(height), Duration::MAX)
                        .await
                        .unwrap();
                assert_eq!(proof.verify_ref(mt.commitment()).unwrap(), leaf.header());
            }
        }

        // Test unknown leaf.
        let err = get_header_proof(&ds, 5, BlockId::Number(4), Duration::from_secs(1))
            .await
            .unwrap_err();
        assert_eq!(err.status(), StatusCode::NOT_FOUND);

        // Test height >= root.
        let err = get_header_proof(&ds, 1, BlockId::Number(1), Duration::MAX)
            .await
            .unwrap_err();
        assert_eq!(err.status(), StatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_namespace_proof() {
        let storage = <DataSource as TestableSequencerDataSource>::create_storage().await;
        let ds = DataSource::create(
            DataSource::persistence_options(&storage),
            Default::default(),
            false,
        )
        .await
        .unwrap();

        // Construct a chain of blocks.
        let client = TestClient::default();
        let leaves = join_all((0..=2).map(|i| client.leaf(i))).await;
        let payloads = join_all((0..=2).map(|i| client.payload(i))).await;
        let vid_commons = join_all((0..=2).map(|i| client.vid_common(i))).await;

        // Save all those objects in the DB.
        {
            let mut tx = ds.write().await.unwrap();
            for (leaf, payload, vid_common) in izip!(&leaves, &payloads, &vid_commons) {
                tx.insert_leaf(leaf).await.unwrap();
                tx.insert_block(&BlockQueryData::<SeqTypes>::new(
                    leaf.header().clone(),
                    payload.clone(),
                ))
                .await
                .unwrap();
                tx.insert_vid(
                    &VidCommonQueryData::<SeqTypes>::new(leaf.header().clone(), vid_common.clone()),
                    None,
                )
                .await
                .unwrap();
            }
            tx.commit().await.unwrap();
        }

        // Test happy path: all blocks.
        let ns = payloads[0]
            .transaction(&TransactionIndex {
                ns_index: 0.into(),
                position: 0,
            })
            .unwrap()
            .namespace();
        let proofs = get_namespace_proof_range(&ds, 0, 3, ns.into(), Duration::MAX, 100)
            .await
            .unwrap();
        assert_eq!(proofs.len(), 3);
        for (leaf, proof) in leaves.iter().zip(proofs) {
            proof.verify(leaf.header(), ns).unwrap();
        }

        // Test happy path: subset.
        let tx = payloads[1]
            .transaction(&TransactionIndex {
                ns_index: 0.into(),
                position: 0,
            })
            .unwrap();
        let ns = tx.namespace();
        let proofs = get_namespace_proof_range(&ds, 1, 2, ns.into(), Duration::MAX, 100)
            .await
            .unwrap();
        assert_eq!(proofs.len(), 1);
        assert_eq!(proofs[0].verify(leaves[1].header(), ns).unwrap(), [tx]);

        // Test missing data in range.
        let err = get_namespace_proof_range(&ds, 0, 4, ns.into(), Duration::from_secs(1), 100)
            .await
            .unwrap_err();
        assert_eq!(err.status(), StatusCode::NOT_FOUND);

        // Test invalid range.
        let err = get_namespace_proof_range(&ds, 1, 0, ns.into(), Duration::from_secs(1), 100)
            .await
            .unwrap_err();
        assert_eq!(err.status(), StatusCode::BAD_REQUEST);
        assert!(
            err.to_string().contains("requested empty interval"),
            "{err:#}"
        );

        // Test large range.
        let err = get_namespace_proof_range(&ds, 0, 10_000, ns.into(), Duration::from_secs(1), 100)
            .await
            .unwrap_err();
        assert_eq!(err.status(), StatusCode::BAD_REQUEST);
        assert!(err.to_string().contains("exceeds maximum size"), "{err:#}");
    }
}