light_client/

client.rs

use std::{collections::HashMap, future::Future, ops::Range};
#[cfg(feature = "client")]
use std::{fmt::Debug, pin::pin, time::Duration};

use anyhow::Result;
#[cfg(feature = "client")]
use anyhow::{Context, anyhow};
#[cfg(feature = "client")]
use derive_builder::Builder;
use espresso_types::{Certificate2, NamespaceId, SeqTypes, v0_3::StakeTableEvent};
#[cfg(feature = "client")]
use futures::{
    FutureExt, TryFuture, TryFutureExt,
    future::{BoxFuture, Either, select, select_ok},
};
#[cfg(feature = "client")]
use hotshot_query_service_types::availability::LeafId;
use hotshot_query_service_types::{availability::LeafQueryData, node::BlockId};
use hotshot_types::data::EpochNumber;
#[cfg(feature = "client")]
use http_client::{StatusCode, Url, error::ClientError as _};
#[cfg(feature = "client")]
use serde::de::DeserializeOwned;
#[cfg(feature = "client")]
use tagged_base64::TaggedBase64;
#[cfg(feature = "client")]
use tokio::time::sleep;
#[cfg(feature = "client")]
use vbs::version::StaticVersion;

use crate::{
    consensus::{
        header::HeaderProof, leaf::LeafProof, namespace::NamespaceProof, payload::PayloadProof,
    },
    storage::LeafRequest,
};

/// `TaggedBase64` tag for the namespace set encoded into the `namespaces` range endpoint URL.
///
/// Shared between the client (which encodes the parameter) and the server (which validates the tag).
pub const NAMESPACES_PARAM_TAG: &str = "NS";

/// Interface to a query server providing the light client API.
pub trait Client: Send + Sync + 'static {
    /// Get the number of blocks known to be in the chain.
    ///
    /// This is equivalent to one more than the block number of the latest known block.
    fn block_height(&self) -> impl Send + Future<Output = Result<u64>>;

    /// Get a finality proof for the requested leaf.
    ///
    /// Optionally, the client may specify the height of a known-finalized leaf. In this case, the
    /// server _may_ terminate the proof in a leaf chain ending at this height, rather than a QC
    /// chain.
    fn leaf_proof(
        &self,
        id: impl Into<LeafRequest> + Send,
        finalized: Option<u64>,
    ) -> impl Send + Future<Output = Result<LeafProof>>;

    /// Get an inclusion proof for the requested header relative to the Merkle tree at height `root`.
    fn header_proof(
        &self,
        root: u64,
        id: BlockId<SeqTypes>,
    ) -> impl Send + Future<Output = Result<HeaderProof>>;

    /// Get all leaves in the given range `[start, end)`.
    fn get_leaves_in_range(
        &self,
        start: usize,
        end: usize,
    ) -> impl Send + Future<Output = Result<Vec<LeafQueryData<SeqTypes>>>>;

    /// Get the leaves the server has in the given ranges, in one request.
    ///
    /// The ranges must be ascending and disjoint. The leaves come without proofs, exactly as
    /// [`get_leaves_in_range`](Self::get_leaves_in_range) does, so the caller must still verify
    /// them.
    fn get_leaves_for_ranges(
        &self,
        ranges: &[Range<u64>],
    ) -> impl Send + Future<Output = Result<Vec<LeafQueryData<SeqTypes>>>>;

    /// Get a proof for the requested payload.
    ///
    /// This method accepts only a `height`, not the more flexible [`BlockId`] type, because a
    /// [`Header`](espresso_types::Header) is needed to verify the resulting proof, so the height
    /// must already be known anyways.
    fn payload_proof(&self, height: u64) -> impl Send + Future<Output = Result<PayloadProof>>;

    /// Get proofs for the payloads in `[start, end)`.
    fn payload_proofs_in_range(
        &self,
        start: u64,
        end: u64,
    ) -> impl Send + Future<Output = Result<Vec<PayloadProof>>>;

    /// Get payload proofs for a set of height ranges, in one request.
    ///
    /// The ranges must be ascending and disjoint, so the proofs come back height-ascending, one
    /// per requested height; callers pair them with leaves positionally.
    fn payload_proofs_for_ranges(
        &self,
        ranges: &[Range<u64>],
    ) -> impl Send + Future<Output = Result<Vec<PayloadProof>>>;

    /// Get a proof for the requested namespace.
    ///
    /// This method accepts only a `height`, not the more flexible [`BlockId`] type, because a
    /// [`Header`](espresso_types::Header) is needed to verify the resulting proof, so the height
    /// must already be known anyways.
    fn namespace_proof(
        &self,
        height: u64,
        namespace: NamespaceId,
    ) -> impl Send + Future<Output = Result<NamespaceProof>>;

    /// Get proofs for the requested namespace for each block in `[start, end)`.
    fn namespace_proofs_in_range(
        &self,
        start: u64,
        end: u64,
        namespace: NamespaceId,
    ) -> impl Send + Future<Output = Result<Vec<NamespaceProof>>>;

    /// Get proofs for the requested namespaces for each block in `[start, end)`.
    fn namespaces_proofs_in_range(
        &self,
        start: u64,
        end: u64,
        namespaces: &[NamespaceId],
    ) -> impl Send + Future<Output = Result<Vec<HashMap<NamespaceId, NamespaceProof>>>>;

    /// Get stake table events for the given epoch.
    ///
    /// This returns the list of events that must be applied to transform the stake table from
    /// `epoch - 1` into the stake table for `epoch`.
    fn stake_table_events(
        &self,
        epoch: EpochNumber,
    ) -> impl Send + Future<Output = Result<Vec<StakeTableEvent>>>;

    /// Get the finality certificate at the given height.
    ///
    /// The returned certificate is unverified. Callers must check the threshold signature
    /// against the appropriate stake table and that it commits to the expected leaf.
    fn cert2(
        &self,
        height: u64,
    ) -> impl Send + Future<Output = Result<Option<Certificate2<SeqTypes>>>>;
}

#[cfg(feature = "client")]
type HttpClient = http_client::Client<hotshot_query_service_types::Error, StaticVersion<0, 1>>;

/// A [`Client`] connected to the HotShot query service.
#[cfg(feature = "client")]
#[derive(Clone, Debug)]
pub struct QueryServiceClient {
    client: HttpClient,
}

#[cfg(feature = "client")]
impl QueryServiceClient {
    /// Connect to a HotShot query service at the given base URL.
    pub fn new(url: Url) -> Self {
        Self {
            client: http_client::Client::builder(url)
                .set_timeout(Some(Duration::from_secs(10)))
                .build(),
        }
    }

    /// GET `path`, deserializing the response
    async fn fetch<T: DeserializeOwned>(&self, path: &str) -> Result<T> {
        self.client
            .get(path)
            .send()
            .await
            .with_context(|| format!("fetching {path}"))
    }
}

#[cfg(feature = "client")]
fn is_not_found(err: &anyhow::Error) -> bool {
    err.chain().any(|cause| {
        cause
            .downcast_ref::<hotshot_query_service_types::Error>()
            .is_some_and(|e| e.status() == StatusCode::NOT_FOUND)
    })
}

#[cfg(feature = "client")]
impl Client for QueryServiceClient {
    async fn block_height(&self) -> Result<u64> {
        self.fetch("/node/block-height").await
    }

    async fn leaf_proof(
        &self,
        id: impl Into<LeafRequest> + Send,
        finalized: Option<u64>,
    ) -> Result<LeafProof> {
        let path = "/light-client/leaf";
        let path = match id.into() {
            LeafRequest::Leaf(LeafId::Number(n)) | LeafRequest::Header(BlockId::Number(n)) => {
                format!("{path}/{n}")
            },
            LeafRequest::Leaf(LeafId::Hash(h)) => format!("{path}/hash/{h}"),
            LeafRequest::Header(BlockId::Hash(h)) => format!("{path}/block-hash/{h}"),
            LeafRequest::Header(BlockId::PayloadHash(h)) => format!("{path}/payload-hash/{h}"),
        };
        let path = match finalized {
            Some(finalized) => format!("{path}/{finalized}"),
            None => path,
        };
        self.fetch(&path).await
    }

    /// Get all leaves in the given range `[start, end)`.
    async fn get_leaves_in_range(
        &self,
        start: usize,
        end: usize,
    ) -> Result<Vec<LeafQueryData<SeqTypes>>> {
        self.fetch(&format!("/availability/leaf/{start}/{end}"))
            .await
    }

    async fn get_leaves_for_ranges(
        &self,
        ranges: &[Range<u64>],
    ) -> Result<Vec<LeafQueryData<SeqTypes>>> {
        self.client
            .post("/availability/leaf/ranges")
            .body_binary(&ranges)?
            .send()
            .await
            .context("fetching leaf ranges")
    }

    async fn header_proof(&self, root: u64, id: BlockId<SeqTypes>) -> Result<HeaderProof> {
        self.fetch(&format!("/light-client/header/{root}/{}", fmt_block_id(id)))
            .await
    }

    async fn payload_proof(&self, height: u64) -> Result<PayloadProof> {
        self.fetch(&format!("/light-client/payload/{height}")).await
    }

    async fn payload_proofs_for_ranges(&self, ranges: &[Range<u64>]) -> Result<Vec<PayloadProof>> {
        self.client
            .post("/light-client/payload/ranges")
            .body_binary(&ranges)?
            .send()
            .await
            .context("fetching payload proof ranges")
    }

    async fn payload_proofs_in_range(&self, start: u64, end: u64) -> Result<Vec<PayloadProof>> {
        self.fetch(&format!("/light-client/payload/{start}/{end}"))
            .await
    }

    async fn namespace_proof(&self, height: u64, namespace: NamespaceId) -> Result<NamespaceProof> {
        self.fetch(&format!("/light-client/namespace/{height}/{namespace}"))
            .await
    }

    async fn namespace_proofs_in_range(
        &self,
        start: u64,
        end: u64,
        namespace: NamespaceId,
    ) -> Result<Vec<NamespaceProof>> {
        self.fetch(&format!(
            "/light-client/namespace/{start}/{end}/{namespace}"
        ))
        .await
    }

    async fn namespaces_proofs_in_range(
        &self,
        start: u64,
        end: u64,
        namespaces: &[NamespaceId],
    ) -> Result<Vec<HashMap<NamespaceId, NamespaceProof>>> {
        let encoded = TaggedBase64::new(NAMESPACES_PARAM_TAG, &serde_json::to_vec(namespaces)?)?;
        self.fetch(&format!("/light-client/namespaces/{start}/{end}/{encoded}"))
            .await
    }

    async fn stake_table_events(&self, epoch: EpochNumber) -> Result<Vec<StakeTableEvent>> {
        self.fetch(&format!("/light-client/stake-table/{epoch}"))
            .await
    }

    async fn cert2(&self, height: u64) -> Result<Option<Certificate2<SeqTypes>>> {
        // A peer that lacks the cert2 responds 404. We surface that as an error rather than
        // `Ok(None)` so `FallbackClient` keeps trying other peers instead of stopping at the first
        // one that lacks it; it resolves absent only once every peer has 404'd.
        //
        // Catchup reads only the peer's storage. `/availability/cert2` would start a fetch on a
        // peer that lacks the cert2, and query nodes peered at each other then keep re-triggering
        // one another's fetches for heights that never had a cert2.
        self.fetch::<Certificate2<SeqTypes>>(&format!("/catchup/{height}/cert2"))
            .await
            .map(Some)
    }
}

#[cfg(feature = "client")]
fn fmt_block_id(id: BlockId<SeqTypes>) -> String {
    match id {
        BlockId::Number(n) => format!("{n}"),
        BlockId::Hash(h) => format!("hash/{h}"),
        BlockId::PayloadHash(h) => format!("payload-hash/{h}"),
    }
}

#[cfg(feature = "client")]
#[derive(Clone, Debug, Builder)]
#[builder(pattern = "owned")]
pub struct FallbackClient<T> {
    /// Duration to wait on one client before starting a fallback request on the next.
    ///
    /// When a fallback request is started, the original request will _not_ be cancelled, and so the
    /// overall operation can complete if the original request completes shortly after the fallback
    /// request is spawned. However, in the event that the original request takes a very long time
    /// or does not complete at all, the fallback allows the overall request to complete without
    /// much delay.
    #[builder(default = Duration::from_millis(300))]
    fallback_delay: Duration,

    /// Ordered list of clients.
    ///
    /// Requests to the [`FallbackClient`] will be tried on each client successively, falling back
    /// down the list until success.
    clients: Vec<T>,
}

#[cfg(feature = "client")]
impl<T> FallbackClient<T> {
    pub fn new(clients: Vec<T>) -> Result<Self, FallbackClientBuilderError> {
        Self::builder().clients(clients).build()
    }

    pub fn builder() -> FallbackClientBuilder<T> {
        Default::default()
    }
}

#[cfg(feature = "client")]
impl<T> Client for FallbackClient<T>
where
    T: Client,
{
    async fn block_height(&self) -> Result<u64> {
        self.get_any(&self.clients, T::block_height).await
    }

    async fn get_leaves_in_range(
        &self,
        start: usize,
        end: usize,
    ) -> Result<Vec<LeafQueryData<SeqTypes>>> {
        self.get_any(&self.clients, |client| {
            client.get_leaves_in_range(start, end)
        })
        .await
    }

    async fn get_leaves_for_ranges(
        &self,
        ranges: &[Range<u64>],
    ) -> Result<Vec<LeafQueryData<SeqTypes>>> {
        self.get_any(&self.clients, |client| client.get_leaves_for_ranges(ranges))
            .await
    }

    async fn header_proof(&self, root: u64, id: BlockId<SeqTypes>) -> Result<HeaderProof> {
        self.get_any(&self.clients, |client| client.header_proof(root, id))
            .await
    }

    async fn leaf_proof(
        &self,
        id: impl Into<LeafRequest> + Send,
        finalized: Option<u64>,
    ) -> Result<LeafProof> {
        let id = id.into();
        self.get_any(&self.clients, |client| client.leaf_proof(id, finalized))
            .await
    }

    async fn namespace_proof(&self, height: u64, namespace: NamespaceId) -> Result<NamespaceProof> {
        self.get_any(&self.clients, |client: &T| {
            client.namespace_proof(height, namespace)
        })
        .await
    }

    async fn namespace_proofs_in_range(
        &self,
        start: u64,
        end: u64,
        namespace: NamespaceId,
    ) -> Result<Vec<NamespaceProof>> {
        self.get_any(&self.clients, |client| {
            client.namespace_proofs_in_range(start, end, namespace)
        })
        .await
    }

    async fn namespaces_proofs_in_range(
        &self,
        start: u64,
        end: u64,
        namespaces: &[NamespaceId],
    ) -> Result<Vec<HashMap<NamespaceId, NamespaceProof>>> {
        self.get_any(&self.clients, |client| {
            client.namespaces_proofs_in_range(start, end, namespaces)
        })
        .await
    }

    async fn payload_proof(&self, height: u64) -> Result<PayloadProof> {
        self.get_any(&self.clients, |client| client.payload_proof(height))
            .await
    }

    async fn payload_proofs_in_range(&self, start: u64, end: u64) -> Result<Vec<PayloadProof>> {
        self.get_any(&self.clients, |client| {
            client.payload_proofs_in_range(start, end)
        })
        .await
    }

    async fn payload_proofs_for_ranges(&self, ranges: &[Range<u64>]) -> Result<Vec<PayloadProof>> {
        self.get_any(&self.clients, |client| {
            client.payload_proofs_for_ranges(ranges)
        })
        .await
    }

    async fn stake_table_events(&self, epoch: EpochNumber) -> Result<Vec<StakeTableEvent>> {
        self.get_any(&self.clients, |client| client.stake_table_events(epoch))
            .await
    }

    async fn cert2(&self, height: u64) -> Result<Option<Certificate2<SeqTypes>>> {
        // Each peer 404s (an error) when it lacks the cert2, so `get_any` keeps trying peers until
        // one has it. Once every peer has 404'd, `get_any` returns that not-found error, which we
        // map back to `Ok(None)` (absent). Any other error propagates so the fetch is retried.
        match self
            .get_any(&self.clients, |client| client.cert2(height))
            .await
        {
            Ok(cert2) => Ok(cert2),
            Err(err) if is_not_found(&err) => Ok(None),
            Err(err) => Err(err),
        }
    }
}

#[cfg(feature = "client")]
impl<T> FallbackClient<T> {
    async fn get_any<C, F>(
        &self,
        clients: impl IntoIterator<Item = C, IntoIter: Send>,
        get: impl Fn(C) -> F + Send,
    ) -> Result<F::Ok>
    where
        C: Send + Sync,
        F: TryFuture<Ok: Send, Error = anyhow::Error> + Send,
    {
        Self::get_any_recursive(0, self.fallback_delay, clients.into_iter(), get, None)
            .await
            .map_err(|err| {
                tracing::warn!(%err, "failed to fetch on all clients");
                err.context("failed to fetch on all clients")
            })
    }

    fn get_any_recursive<'a, C, F>(
        i: usize,
        timeout: Duration,
        mut clients: impl Iterator<Item = C> + Send + 'a,
        get: impl Fn(C) -> F + Send + 'a,
        prev_err: Option<anyhow::Error>,
    ) -> BoxFuture<'a, Result<F::Ok>>
    where
        C: Send + Sync + 'a,
        F: TryFuture<Ok: Send, Error = anyhow::Error> + Send + 'a,
    {
        async move {
            let Some(client) = clients.next() else {
                return Err(prev_err.unwrap_or_else(|| anyhow!("no clients available")));
            };
            match select(
                pin!(TryFutureExt::into_future(get(client))),
                pin!(sleep(timeout)),
            )
            .await
            {
                Either::Left((Ok(res), _)) => {
                    tracing::debug!("fetch succeeded on client {i}");
                    Ok(res)
                },
                Either::Left((Err(err), _)) => {
                    // If the pending future fails, abandon it and immediately start the fallback
                    // even if we haven't hit the timeout.
                    tracing::info!(i, %err, "failed to fetch on client");
                    Self::get_any_recursive(i + 1, timeout, clients, get, Some(err)).await
                },
                Either::Right((_, fut)) => {
                    // The timeout has elapsed. Start the request on the next client, but also keep
                    // polling the in-progress future from the first request, since there is still a
                    // chance that it will complete before the fallback request.
                    tracing::info!(
                        "fetch on client {i} took longer than {timeout:?}, starting fallback \
                         request"
                    );
                    Ok(select_ok([
                        fut.boxed(),
                        Self::get_any_recursive(i + 1, timeout, clients, get, prev_err),
                    ])
                    .await?
                    .0)
                },
            }
        }
        .boxed()
    }
}

#[cfg(test)]
mod test {
    use std::{marker::PhantomData, time::Duration};

    use committable::Committable;
    use espresso_node::{
        api::{
            Options,
            data_source::testing::TestableSequencerDataSource,
            sql,
            test_helpers::{TestNetwork, TestNetworkConfigBuilder},
        },
        testing::{TestConfigBuilder, wait_for_decide_on_handle},
    };
    use espresso_types::{Header, TEST_UPGRADE, Transaction};
    use futures::{TryStreamExt, stream::StreamExt};
    use hotshot_query_service_types::{
        Resolvable,
        availability::{BlockQueryData, LeafQueryData},
    };
    use pretty_assertions::assert_eq;
    use rand::RngCore;
    use rstest_reuse::apply;
    use test_utils;
    use tokio::time::sleep;

    use super::*;
    use crate::{
        consensus::leaf::{FinalityProof, LeafProofHint},
        testing::AlwaysTrueQuorum,
    };

    #[rstest_reuse::template]
    #[rstest::rstest]
    #[case::postgres(PhantomData::<sql::postgres::DataSource>)]
    #[case::sqlite(PhantomData::<sql::sqlite::DataSource>)]
    #[tokio::test]
    #[test_log::test]
    fn sql_data_sources<D: TestableSequencerDataSource>(#[case] _d: PhantomData<D>) {}

    fn client(url: Url) -> impl Client {
        // Run all tests with a fallback client where the first client is an invalid URL, to stress
        // test failover.
        FallbackClient::new(vec![
            QueryServiceClient::new("http://notarealurl".parse().unwrap()),
            QueryServiceClient::new(url),
        ])
        .unwrap()
    }

    #[apply(sql_data_sources)]
    async fn test_block_height<D: TestableSequencerDataSource>(_d: PhantomData<D>) {
        let port =
            test_utils::reserve_tcp_port().expect("OS should have ephemeral ports available");
        let url: Url = format!("http://localhost:{port}").parse().unwrap();

        let test_config = TestConfigBuilder::default().build();
        let storage = D::create_storage().await;
        let persistence = <D as TestableSequencerDataSource>::persistence_options(&storage);

        let config = TestNetworkConfigBuilder::<1, _, _>::with_num_nodes()
            .api_config(D::options(&storage, Options::with_port(port)))
            .persistences([persistence])
            .network_config(test_config)
            .build();

        let _network = TestNetwork::new(config, TEST_UPGRADE).await;
        let client = client(url);

        // Check that the block height increases over time.
        let initial_height = client.block_height().await.unwrap();
        tracing::info!(initial_height);

        loop {
            sleep(Duration::from_secs(1)).await;
            let height = client.block_height().await.unwrap();
            if height > initial_height {
                tracing::info!(height, initial_height, "height increased");
                break;
            }
            tracing::info!("waiting for height to increase");
        }
    }

    #[apply(sql_data_sources)]
    async fn test_leaf_proof<D: TestableSequencerDataSource>(_d: PhantomData<D>) {
        let port =
            test_utils::reserve_tcp_port().expect("OS should have ephemeral ports available");
        let url: Url = format!("http://localhost:{port}").parse().unwrap();

        let test_config = TestConfigBuilder::default().build();
        let epoch_height = test_config.hotshot_config().epoch_height;
        let storage = D::create_storage().await;
        let persistence = <D as TestableSequencerDataSource>::persistence_options(&storage);

        let config = TestNetworkConfigBuilder::<1, _, _>::with_num_nodes()
            .api_config(D::options(&storage, Options::with_port(port)))
            .persistences([persistence])
            .network_config(test_config)
            .build();

        let _network = TestNetwork::new(config, TEST_UPGRADE).await;
        let client = client(url.clone());

        // Wait for a chain of leaves to be produced.
        let leaves: Vec<LeafQueryData<SeqTypes>> = HttpClient::new(url)
            .socket("availability/stream/leaves/1")
            .subscribe()
            .await
            .unwrap()
            .take(2)
            .try_collect()
            .await
            .unwrap();

        // Get leaf proof by height.
        let proof = client.leaf_proof(LeafId::Number(1), Some(2)).await.unwrap();
        assert!(matches!(proof.proof(), FinalityProof::Assumption));
        assert_eq!(
            proof
                .verify(LeafProofHint::assumption(leaves[1].leaf()))
                .await
                .unwrap(),
            leaves[0]
        );

        // Get the same proof by various other IDs.
        for req in [
            LeafRequest::Header(BlockId::Number(1)),
            LeafRequest::Leaf(LeafId::Hash(leaves[0].hash())),
            LeafRequest::Header(BlockId::Hash(leaves[0].block_hash())),
        ] {
            tracing::info!(?req, "get proof by alternative ID");
            let proof = client.leaf_proof(req, None).await.unwrap();
            assert!(matches!(proof.proof(), FinalityProof::NewProtocol { .. }));
            assert_eq!(
                proof
                    .verify(LeafProofHint::Quorum(&AlwaysTrueQuorum::new(epoch_height)))
                    .await
                    .unwrap(),
                leaves[0]
            );
        }

        // Get a proof by payload hash (this doesn't necessarily return a unique leaf, since
        // multiple) leaves may have the same payload.
        let proof = client
            .leaf_proof(BlockId::PayloadHash(leaves[0].payload_hash()), None)
            .await
            .unwrap();
        assert_eq!(
            proof
                .verify(LeafProofHint::Quorum(&AlwaysTrueQuorum::new(epoch_height)))
                .await
                .unwrap()
                .payload_hash(),
            leaves[0].payload_hash()
        );

        // Get a range of leaves.
        assert_eq!(client.get_leaves_in_range(1, 3).await.unwrap(), leaves);
    }

    #[apply(sql_data_sources)]
    async fn test_header_proof<D: TestableSequencerDataSource>(_d: PhantomData<D>) {
        let port =
            test_utils::reserve_tcp_port().expect("OS should have ephemeral ports available");
        let url: Url = format!("http://localhost:{port}").parse().unwrap();

        let test_config = TestConfigBuilder::default().build();
        let storage = D::create_storage().await;
        let persistence = <D as TestableSequencerDataSource>::persistence_options(&storage);

        let config = TestNetworkConfigBuilder::<1, _, _>::with_num_nodes()
            .api_config(D::options(&storage, Options::with_port(port)))
            .persistences([persistence])
            .network_config(test_config)
            .build();

        let _network = TestNetwork::new(config, TEST_UPGRADE).await;
        let client = client(url.clone());
        let http = HttpClient::new(url);

        // Wait for a chain of blocks to be produced.
        let headers: Vec<Header> = http
            .socket("availability/stream/headers/1")
            .subscribe()
            .await
            .unwrap()
            .take(2)
            .try_collect()
            .await
            .unwrap();
        // Wait for the state API to catch up.
        loop {
            let state_height: u64 = http.get("block-state/block-height").send().await.unwrap();
            if state_height >= 2 {
                break;
            }
            tracing::info!(state_height, "waiting for block state to reach height 2");
            sleep(Duration::from_secs(1)).await;
        }

        // Get header proof by height.
        let proof = client.header_proof(2, BlockId::Number(1)).await.unwrap();
        assert_eq!(
            proof.verify(headers[1].block_merkle_tree_root()).unwrap(),
            headers[0]
        );

        // Get the same proof by block hash.
        let proof = client
            .header_proof(2, BlockId::Hash(headers[0].commitment()))
            .await
            .unwrap();
        assert_eq!(
            proof.verify(headers[1].block_merkle_tree_root()).unwrap(),
            headers[0]
        );

        // Get a proof by payload hash (this doesn't necessarily return a unique header, since
        // multiple headers may have the same payload).
        let proof = client
            .header_proof(2, BlockId::PayloadHash(headers[0].payload_commitment()))
            .await
            .unwrap();
        assert_eq!(
            proof
                .verify(headers[1].block_merkle_tree_root())
                .unwrap()
                .payload_commitment(),
            headers[0].payload_commitment()
        );
    }

    #[apply(sql_data_sources)]
    async fn test_payload_proof<D: TestableSequencerDataSource>(_d: PhantomData<D>) {
        let port =
            test_utils::reserve_tcp_port().expect("OS should have ephemeral ports available");
        let url: Url = format!("http://localhost:{port}").parse().unwrap();

        let test_config = TestConfigBuilder::default().build();
        let storage = D::create_storage().await;
        let persistence = <D as TestableSequencerDataSource>::persistence_options(&storage);

        let config = TestNetworkConfigBuilder::<1, _, _>::with_num_nodes()
            .api_config(D::options(&storage, Options::with_port(port)))
            .persistences([persistence])
            .network_config(test_config)
            .build();

        let _network = TestNetwork::new(config, TEST_UPGRADE).await;
        let client = client(url.clone());

        // Wait for a few blocks to be produced.
        let blocks: Vec<BlockQueryData<SeqTypes>> = HttpClient::new(url)
            .socket("availability/stream/blocks/1")
            .subscribe()
            .await
            .unwrap()
            .take(2)
            .try_collect()
            .await
            .unwrap();

        let proof = client.payload_proof(1).await.unwrap();
        assert_eq!(
            proof.verify(blocks[0].header()).unwrap(),
            *blocks[0].payload()
        );

        // The block will be empty, but check at least that the namespace proof method hits the
        // correct endpoint and deserializes correctly.
        let ns = NamespaceId::from(0u64);
        let ns_proof = client.namespace_proof(1, ns).await.unwrap();
        assert_eq!(ns_proof.verify(blocks[0].header(), ns).unwrap(), vec![]);

        // Get a range of payloads.
        let proofs = client.payload_proofs_in_range(1, 3).await.unwrap();
        assert_eq!(proofs.len(), 2);
        for (block, proof) in blocks.iter().zip(proofs) {
            assert_eq!(proof.verify(block.header()).unwrap(), *block.payload());
        }
    }

    #[apply(sql_data_sources)]
    async fn test_namespace_proof<D: TestableSequencerDataSource>(_d: PhantomData<D>) {
        let port =
            test_utils::reserve_tcp_port().expect("OS should have ephemeral ports available");
        let url: Url = format!("http://localhost:{port}").parse().unwrap();

        let test_config = TestConfigBuilder::default().build();
        let storage = D::create_storage().await;
        let persistence = <D as TestableSequencerDataSource>::persistence_options(&storage);

        let config = TestNetworkConfigBuilder::<1, _, _>::with_num_nodes()
            .api_config(D::options(&storage, Options::with_port(port)))
            .persistences([persistence])
            .network_config(test_config)
            .build();

        let network = TestNetwork::new(config, TEST_UPGRADE).await;
        let client = client(url.clone());
        let http = HttpClient::new(url);

        // Submit a couple of transactions to form non-empty blocks.
        let ns = NamespaceId::from(1u64);
        let mut events = network.server.event_stream();
        let mut txs = vec![];
        let mut headers = vec![];
        for _ in 0..2 {
            let mut bytes = vec![0; 32];
            rand::thread_rng().fill_bytes(&mut bytes);
            let tx = Transaction::new(ns, bytes);
            network.server.submit_transaction(tx.clone()).await.unwrap();
            let block = wait_for_decide_on_handle(&mut events, &tx).await.0;
            tracing::info!(block, hash = %tx.commit(), ?tx, "transaction included");

            let header: Header = http
                .get(&format!("availability/header/{block}"))
                .send()
                .await
                .unwrap();

            txs.push(tx);
            headers.push(header);
        }

        // Query namespaces individually.
        for (tx, header) in txs.iter().zip(&headers) {
            let proof = client.namespace_proof(header.height(), ns).await.unwrap();
            assert_eq!(proof.verify(header, ns).unwrap(), std::slice::from_ref(tx));
        }

        // Query both namespaces by range.
        let proofs = client
            .namespace_proofs_in_range(headers[0].height(), headers[1].height() + 1, ns)
            .await
            .unwrap();
        assert_eq!(
            proofs.len() as u64,
            headers[1].height() + 1 - headers[0].height()
        );
        assert_eq!(
            proofs[0].verify(&headers[0], ns).unwrap(),
            std::slice::from_ref(&txs[0])
        );
        assert_eq!(
            proofs[proofs.len() - 1].verify(&headers[1], ns).unwrap(),
            std::slice::from_ref(&txs[1])
        );
        // All other blocks in the range should be empty.
        for (i, proof) in proofs.iter().enumerate().take(proofs.len() - 1).skip(1) {
            let header = http
                .get(&format!(
                    "availability/header/{}",
                    headers[0].height() + (i as u64)
                ))
                .send()
                .await
                .unwrap();
            assert_eq!(proof.verify(&header, ns).unwrap(), vec![]);
        }

        let absent = NamespaceId::from(u64::from(ns) + 1);
        let start = headers[0].height();
        let end = headers[1].height() + 1;
        let blocks = client
            .namespaces_proofs_in_range(start, end, &[ns, absent])
            .await
            .unwrap();
        assert_eq!(blocks.len() as u64, end - start);

        for (i, block) in blocks.iter().enumerate() {
            let header: Header = http
                .get(&format!("availability/header/{}", start + i as u64))
                .send()
                .await
                .unwrap();
            for queried in [ns, absent] {
                match (header.ns_table().find_ns_id(&queried), block.get(&queried)) {
                    (Some(_), Some(proof)) => {
                        proof.verify(&header, queried).unwrap();
                    },
                    (None, None) => {},
                    (present, returned) => panic!(
                        "block {} namespace {queried}: header present={}, returned={}",
                        start + i as u64,
                        present.is_some(),
                        returned.is_some()
                    ),
                }
            }
        }

        // The first and last blocks are the ones we submitted `ns` transactions to.
        assert_eq!(
            blocks[0].get(&ns).unwrap().verify(&headers[0], ns).unwrap(),
            std::slice::from_ref(&txs[0])
        );
        assert_eq!(
            blocks[blocks.len() - 1]
                .get(&ns)
                .unwrap()
                .verify(&headers[1], ns)
                .unwrap(),
            std::slice::from_ref(&txs[1])
        );
    }
}