espresso_node/api/endpoints/

availability.rs

use std::time::Duration;

use espresso_types::{NamespaceId, NsProof, PubKey, StateCertQueryDataV1, StateCertQueryDataV2};
use futures::{
    future::{FutureExt, TryFutureExt, try_join_all},
    join,
    stream::{Stream, StreamExt, TryStreamExt},
    try_join,
};
use hotshot_query_service::{
    ApiState,
    availability::{
        self, AvailabilityDataSource, BlockQueryData, Error, FetchBlockSnafu, VidCommonQueryData,
    },
    node::{BlockId, NodeDataSource},
    types::HeightIndexed,
};
use hotshot_types::{
    data::VidShare, simple_certificate::LightClientStateUpdateCertificateV2,
    traits::network::ConnectedNetwork, vid::avidm::AvidMShare,
};
use snafu::OptionExt;
use tide_disco::{Api, RequestParams, StatusCode, method::ReadState};
use tracing::warn;
use vbs::version::StaticVersionType;

use crate::{
    SeqTypes, SequencerApiVersion, SequencerPersistence,
    api::{
        StorageState,
        data_source::{
            RequestResponseDataSource, SequencerDataSource, StateCertDataSource,
            StateCertFetchingDataSource,
        },
    },
};

pub(in crate::api) type AvailState<N, P, D> = ApiState<StorageState<N, P, D>>;

type AvailabilityApi<N, P, D, ApiVer> = Api<AvailState<N, P, D>, Error, ApiVer>;

/// Get a namespace proof for the given block, if possible.
///
/// Always returns the newest supported proof type, which supports the greatest number of possible
/// cases (e.g. proofs can still be generated even if the block was maliciously encoded). For
/// backwards compatibility, the resulting proof can be downgraded. However, this may fail in case a
/// proof was only able to be generated by making use of one of the newer proof types.
///
/// Returns no proof (`Ok(None)`) if the requested namespace is not present at all in the given
/// block.
async fn get_namespace_proof<S>(
    block: &BlockQueryData<SeqTypes>,
    common: &VidCommonQueryData<SeqTypes>,
    ns_id: NamespaceId,
    state: &S,
) -> Result<Option<NsProof>, Error>
where
    S: ReadState,
    S::State: NodeDataSource<SeqTypes> + RequestResponseDataSource<SeqTypes> + Sync,
{
    let ns_table = block.payload().ns_table();
    let Some(ns_index) = ns_table.find_ns_id(&ns_id) else {
        return Ok(None);
    };

    // Optimistically, try to generate a
    // proof for a correctly encoded block.
    if let Some(proof) = NsProof::new(block.payload(), &ns_index, common.common()) {
        return Ok(Some(proof));
    }

    // If we fail to generate the correct encoding proof, try to generate a v1.1 proof, which
    // supports proof of incorrect encoding.
    tracing::warn!(
        height = block.height(),
        ?ns_id,
        "Failed to generate namespace proof, trying to generate incorrect encoding proof"
    );
    let vid_shares_req = state
        .read(move |state| {
            state
                .request_vid_shares(block.height(), common.clone(), Duration::from_secs(40))
                .boxed()
        })
        .await;
    let mut vid_shares = vid_shares_req.await.map_err(|err| {
        warn!("Failed to request VID shares from network: {err:#}");
        hotshot_query_service::availability::Error::Custom {
            message: "Failed to request VID shares from network".to_string(),
            status: StatusCode::NOT_FOUND,
        }
    })?;
    let vid_share = state
        .read(|state| state.vid_share(block.height() as usize).boxed())
        .await;
    if let Ok(vid_share) = vid_share {
        vid_shares.push(vid_share);
    };

    // Collect the shares as V1 shares
    let vid_shares: Vec<AvidMShare> = vid_shares
        .into_iter()
        .filter_map(|share| {
            if let VidShare::V1(share) = share {
                Some(share)
            } else {
                None
            }
        })
        .collect();

    if let Some(proof) = NsProof::v1_1_new_with_incorrect_encoding(
        &vid_shares,
        ns_table,
        &ns_index,
        &common.payload_hash(),
        common.common(),
    ) {
        return Ok(Some(proof));
    }

    Err(Error::Custom {
        message: "Failed to generate proof of incorrect encoding".to_string(),
        status: StatusCode::INTERNAL_SERVER_ERROR,
    })
}

fn extract_ns_proof_v1(
    proof: Option<NsProof>,
    ns_id: NamespaceId,
) -> Result<espresso_types::NamespaceProofQueryData, Error> {
    let transactions = proof
        .as_ref()
        .map(|proof| proof.export_all_txs(&ns_id))
        .unwrap_or_default();
    Ok(espresso_types::NamespaceProofQueryData {
        transactions,
        proof,
    })
}

fn extract_ns_proof_v0(
    proof: Option<NsProof>,
    ns_id: NamespaceId,
) -> Result<espresso_types::ADVZNamespaceProofQueryData, Error> {
    let proof = match proof {
        Some(NsProof::V0(proof)) => Some(proof),
        Some(_) => {
            return Err(Error::Custom {
                message: "Unsupported VID version, use new API version instead.".to_string(),
                status: StatusCode::NOT_FOUND,
            });
        },
        None => None,
    };
    let transactions = proof
        .as_ref()
        .map(|proof| proof.export_all_txs(&ns_id))
        .unwrap_or_default();
    Ok(espresso_types::ADVZNamespaceProofQueryData {
        transactions,
        proof,
    })
}

async fn get_block_for_ns_proof<S>(
    req: &RequestParams,
    state: &S,
    timeout: Duration,
) -> Result<(BlockQueryData<SeqTypes>, VidCommonQueryData<SeqTypes>), Error>
where
    S: ReadState,
    S::State: AvailabilityDataSource<SeqTypes> + Sync,
{
    let id = if let Some(height) = req.opt_integer_param("height")? {
        BlockId::Number(height)
    } else if let Some(hash) = req.opt_blob_param("hash")? {
        BlockId::Hash(hash)
    } else {
        BlockId::PayloadHash(req.blob_param("payload-hash")?)
    };
    let (fetch_block, fetch_vid) = state
        .read(|state| async move { join!(state.get_block(id), state.get_vid_common(id)) }.boxed())
        .await;
    try_join!(
        async move {
            fetch_block
                .with_timeout(timeout)
                .await
                .context(FetchBlockSnafu {
                    resource: id.to_string(),
                })
        },
        async move {
            fetch_vid
                .with_timeout(timeout)
                .await
                .context(FetchBlockSnafu {
                    resource: id.to_string(),
                })
        }
    )
}

async fn get_block_range_for_ns_proof<S>(
    req: &RequestParams,
    state: &S,
    limit: usize,
    timeout: Duration,
) -> Result<Vec<(BlockQueryData<SeqTypes>, VidCommonQueryData<SeqTypes>)>, Error>
where
    S: ReadState,
    S::State: AvailabilityDataSource<SeqTypes> + Sync,
{
    let from: usize = req.integer_param("from")?;
    let until: usize = req.integer_param("until")?;
    if until.saturating_sub(from) > limit {
        return Err(Error::RangeLimit { from, until, limit });
    }

    let (blocks, vids) = state
        .read(|state| {
            async move {
                join!(
                    state.get_block_range(from..until),
                    state.get_vid_common_range(from..until)
                )
            }
            .boxed()
        })
        .await;
    blocks
        .zip(vids)
        .enumerate()
        .then(|(i, (block, vid))| async move {
            let (Some(block), Some(vid)) =
                join!(block.with_timeout(timeout), vid.with_timeout(timeout),)
            else {
                return Err(Error::FetchBlock {
                    resource: (from + i).to_string(),
                });
            };
            Ok((block, vid))
        })
        .try_collect()
        .await
}

fn get_block_stream_for_ns_proof<'a, S>(
    req: RequestParams,
    state: &'a S,
) -> impl 'a
+ Stream<
    Item = Result<
        (
            NamespaceId,
            BlockQueryData<SeqTypes>,
            VidCommonQueryData<SeqTypes>,
        ),
        Error,
    >,
>
where
    S: ReadState,
    S::State: AvailabilityDataSource<SeqTypes> + Sync,
{
    async move {
        let ns_id = NamespaceId::from(req.integer_param::<_, u32>("namespace")?);
        let height = req.integer_param("height")?;
        Ok(state
            .read(|state| {
                async move {
                    state
                        .subscribe_blocks(height)
                        .await
                        .zip(state.subscribe_vid_common(height).await)
                        .map(move |(block, vid)| (ns_id, block, vid))
                        .map(Ok)
                }
                .boxed()
            })
            .await)
    }
    .try_flatten_stream()
}

async fn get_state_cert<S>(
    state: &S,
    epoch: u64,
    timeout: Duration,
) -> Result<LightClientStateUpdateCertificateV2<SeqTypes>, availability::Error>
where
    S: ReadState,
    S::State: StateCertDataSource + StateCertFetchingDataSource<SeqTypes> + Sync,
{
    // Try to get from local storage first
    let state_cert = state
        .read(|state| state.get_state_cert_by_epoch(epoch).boxed())
        .await
        .map_err(|e| availability::Error::Custom {
            message: format!("Failed to get state cert: {e}"),
            status: StatusCode::INTERNAL_SERVER_ERROR,
        })?;

    match state_cert {
        Some(cert) => Ok(cert),
        None => {
            // Not found locally, try to fetch from peers
            let cert = state
                .read(|state| state.request_state_cert(epoch, timeout).boxed())
                .await?;

            // Store the fetched certificate
            state
                .read(|state| state.insert_state_cert(epoch, cert.clone()).boxed())
                .await
                .map_err(|e| availability::Error::Custom {
                    message: format!("Failed to store state cert: {e}"),
                    status: StatusCode::INTERNAL_SERVER_ERROR,
                })?;

            Ok(cert)
        },
    }
}

// TODO (abdul): replace snafu with `this_error` in  hotshot query service
// Snafu has been replaced by `this_error` everywhere.
// However, the query service still uses snafu
pub(in crate::api) fn availability<N, P, D>(
    api_ver: semver::Version,
) -> anyhow::Result<AvailabilityApi<N, P, D, SequencerApiVersion>>
where
    N: ConnectedNetwork<PubKey>,
    D: SequencerDataSource + Send + Sync + 'static,
    P: SequencerPersistence,
{
    let mut options = availability::Options::default();
    let extension = toml::from_str(include_str!("../../../api/availability.toml"))?;
    options.extensions.push(extension);
    let timeout = options.fetch_timeout;
    let limit = options.large_object_range_limit;

    let mut api = availability::define_api::<AvailState<N, P, D>, SeqTypes, _>(
        &options,
        SequencerApiVersion::instance(),
        api_ver.clone(),
    )?;

    if api_ver.major == 1 {
        api.at("getnamespaceproof", move |req, state| {
            async move {
                let ns_id = NamespaceId::from(req.integer_param::<_, u32>("namespace")?);
                let (block, common) = get_block_for_ns_proof(&req, state, timeout).await?;
                let proof = get_namespace_proof(&block, &common, ns_id, state).await?;
                extract_ns_proof_v1(proof, ns_id)
            }
            .boxed()
        })?
        .at("getnamespaceproof_range", move |req, state| {
            async move {
                let ns_id = NamespaceId::from(req.integer_param::<_, u32>("namespace")?);
                let blocks = get_block_range_for_ns_proof(&req, state, limit, timeout).await?;
                try_join_all(blocks.iter().map(|(block, vid)| async move {
                    let proof = get_namespace_proof(block, vid, ns_id, state).await?;
                    extract_ns_proof_v1(proof, ns_id)
                }))
                .await
            }
            .boxed()
        })?
        .stream("stream_namespace_proofs", move |req, state| {
            get_block_stream_for_ns_proof(req, state)
                .and_then(move |(ns_id, block, vid)| async move {
                    let proof = get_namespace_proof(&block, &vid, ns_id, state).await?;
                    extract_ns_proof_v1(proof, ns_id)
                })
                .boxed()
        })?;
    } else if api_ver.major == 0 {
        api.at("getnamespaceproof", move |req, state| {
            async move {
                let ns_id = NamespaceId::from(req.integer_param::<_, u32>("namespace")?);
                let (block, common) = get_block_for_ns_proof(&req, state, timeout).await?;
                let proof = get_namespace_proof(&block, &common, ns_id, state).await?;
                extract_ns_proof_v0(proof, ns_id)
            }
            .boxed()
        })?
        .at("getnamespaceproof_range", move |req, state| {
            async move {
                let ns_id = NamespaceId::from(req.integer_param::<_, u32>("namespace")?);
                let blocks = get_block_range_for_ns_proof(&req, state, limit, timeout).await?;
                try_join_all(blocks.iter().map(|(block, vid)| async move {
                    let proof = get_namespace_proof(block, vid, ns_id, state).await?;
                    extract_ns_proof_v0(proof, ns_id)
                }))
                .await
            }
            .boxed()
        })?
        .stream("stream_namespace_proofs", move |req, state| {
            get_block_stream_for_ns_proof(req, state)
                .and_then(move |(ns_id, block, vid)| async move {
                    let proof = get_namespace_proof(&block, &vid, ns_id, state).await?;
                    extract_ns_proof_v0(proof, ns_id)
                })
                .boxed()
        })?;
    }

    if api_ver.major >= 1 {
        api.at("incorrect_encoding_proof", move |req, state| {
            async move {
                let ns_id = NamespaceId::from(req.integer_param::<_, u32>("namespace")?);
                let (block, common) = get_block_for_ns_proof(&req, state, timeout).await?;
                match get_namespace_proof(&block, &common, ns_id, state).await? {
                    Some(NsProof::V1IncorrectEncoding(proof)) => Ok(proof),
                    Some(_) => Err(Error::Custom {
                        message: "block was correctly encoded".into(),
                        status: StatusCode::NOT_FOUND,
                    }),
                    None => Err(Error::Custom {
                        message: "namespace not present in block".into(),
                        status: StatusCode::NOT_FOUND,
                    }),
                }
            }
            .boxed()
        })?;
    }

    api.at("get_state_cert", move |req, state| {
        async move {
            let epoch: u64 = req.integer_param("epoch")?;
            let cert = get_state_cert(state, epoch, timeout).await?;
            Ok(StateCertQueryDataV1::from(StateCertQueryDataV2(cert)))
        }
        .boxed()
    })?;

    api.at("get_state_cert_v2", move |req, state| {
        async move {
            let epoch: u64 = req.integer_param("epoch")?;
            let cert = get_state_cert(state, epoch, timeout).await?;
            Ok(StateCertQueryDataV2(cert))
        }
        .boxed()
    })?;

    Ok(api)
}