diff --git a/Cargo.lock b/Cargo.lock index 91216a9fbc5..3211c116f23 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4844,6 +4844,7 @@ dependencies = [ "csv", "derivative", "derive_more 2.1.1", + "disco-types", "dotenvy", "dyn-clone", "either", @@ -4897,7 +4898,6 @@ dependencies = [ "reqwest 0.12.28", "rstest", "rstest_reuse", - "semver 1.0.28", "serde", "serde_bytes", "serde_json", @@ -4913,7 +4913,6 @@ dependencies = [ "test-log", "test-utils", "thiserror 2.0.18", - "tide-disco", "time 0.3.51", "tokio", "tokio-tungstenite 0.28.0", @@ -6533,6 +6532,7 @@ dependencies = [ "blake3", "clap 4.6.1", "csv", + "disco-types", "futures", "hotshot-types", "http-client 0.1.0", @@ -6540,7 +6540,6 @@ dependencies = [ "multiaddr", "serde", "serde_json", - "tide-disco", "tokio", "toml", "tracing", @@ -8697,6 +8696,7 @@ dependencies = [ "anyhow", "axum 0.8.9", "clap 4.6.1", + "disco-types", "espresso-node", "espresso-types", "futures", @@ -8707,7 +8707,6 @@ dependencies = [ "log-panics", "serde", "serde_json", - "tide-disco", "tokio", "toml", "tracing", diff --git a/crates/espresso/node/Cargo.toml b/crates/espresso/node/Cargo.toml index 3aeb5baa2e2..0aa96fa040b 100644 --- a/crates/espresso/node/Cargo.toml +++ b/crates/espresso/node/Cargo.toml @@ -50,6 +50,7 @@ committable = { workspace = true } csv = { workspace = true } derivative = { workspace = true } derive_more = { workspace = true } +disco-types = { workspace = true } dotenvy = { workspace = true } dyn-clone = { workspace = true } either = { workspace = true } @@ -97,7 +98,6 @@ request-response = { workspace = true } reqwest = { workspace = true } rstest = { workspace = true } rstest_reuse = { workspace = true } -semver = { workspace = true } serde = { workspace = true } serde_bytes = { workspace = true } serde_json = { workspace = true } @@ -112,7 +112,6 @@ tagged-base64 = { workspace = true } tempfile = { workspace = true } test-utils = { workspace = true } thiserror = { workspace = true } -tide-disco = { workspace = true } time = { workspace = true } tokio = { workspace = true } tokio-tungstenite = { workspace = true } diff --git a/crates/espresso/node/src/api.rs b/crates/espresso/node/src/api.rs index 7253b4ec623..3025578dfab 100644 --- a/crates/espresso/node/src/api.rs +++ b/crates/espresso/node/src/api.rs @@ -90,7 +90,6 @@ use crate::{ }; pub mod data_source; -pub mod endpoints; pub mod fs; pub mod light_client; pub mod options; @@ -1883,10 +1882,7 @@ pub mod test_helpers { MOCK_SEQUENCER_VERSIONS, NamespaceId, ValidatedState, v0::traits::{NullEventConsumer, PersistenceOptions, StateCatchup}, }; - use futures::{ - future::{FutureExt, join_all}, - stream::StreamExt, - }; + use futures::{future::join_all, stream::StreamExt}; use hotshot::types::{Event, EventType}; use hotshot_contract_adapter::stake_table::StakeTableContractVersion; use hotshot_types::{ @@ -1899,10 +1895,8 @@ pub mod test_helpers { use staking_cli::demo::{DelegationConfig, StakingTransactions}; use tempfile::TempDir; use test_utils::reserve_tcp_port; - use tide_disco::{Api, App, Error, StatusCode}; - use tokio::{spawn, task::JoinHandle, time::sleep}; - use url::Url; - use vbs::version::{StaticVersion, StaticVersionType}; + use tokio::time::sleep; + use vbs::version::StaticVersion; use versions::{EPOCH_VERSION, Upgrade}; use super::*; @@ -2524,70 +2518,6 @@ pub mod test_helpers { .unwrap() .unwrap(); } - - pub async fn spawn_dishonest_peer_catchup_api() -> anyhow::Result<(Url, JoinHandle<()>)> { - let toml = toml::from_str::(include_str!("../api/catchup.toml")).unwrap(); - let mut api = - Api::<(), hotshot_query_service::Error, SequencerApiVersion>::new(toml).unwrap(); - - api.get("account", |_req, _state: &()| { - async move { - Result::::Err(hotshot_query_service::Error::catch_all( - StatusCode::BAD_REQUEST, - "no account found".to_string(), - )) - } - .boxed() - })? - .get("blocks", |_req, _state| { - async move { - Result::::Err(hotshot_query_service::Error::catch_all( - StatusCode::BAD_REQUEST, - "no block found".to_string(), - )) - } - .boxed() - })? - .get("chainconfig", |_req, _state| { - async move { - Result::::Ok(ChainConfig { - max_block_size: 300.into(), - base_fee: 1.into(), - fee_recipient: "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48" - .parse() - .unwrap(), - ..Default::default() - }) - } - .boxed() - })? - .get("leafchain", |_req, _state| { - async move { - Result::, _>::Err(hotshot_query_service::Error::catch_all( - StatusCode::BAD_REQUEST, - "No leafchain found".to_string(), - )) - } - .boxed() - })?; - - let mut app = App::<_, hotshot_query_service::Error>::with_state(()); - app.with_version(env!("CARGO_PKG_VERSION").parse().unwrap()); - - app.register_module::<_, _>("catchup", api).unwrap(); - - let port = reserve_tcp_port().expect("OS should have ephemeral ports available"); - let url: Url = Url::parse(&format!("http://localhost:{port}")).unwrap(); - - let handle = spawn({ - let url = url.clone(); - async move { - let _ = app.serve(url, SequencerApiVersion::instance()).await; - } - }); - - Ok((url, handle)) - } } #[cfg(test)] @@ -3242,7 +3172,11 @@ mod test { utils::epoch_from_block_number, x25519, }; - use http_client::{Client, StatusCode, error::ClientErr}; + use http_client::{ + Client, StatusCode, + error::ClientErr, + healthcheck::{AppHealth, HealthStatus}, + }; use jf_merkle_tree_compat::{ MerkleTreeScheme, prelude::{MerkleProof, Sha3Node}, @@ -3259,7 +3193,6 @@ mod test { status_test_helper, submit_test_helper, }; use test_utils::reserve_tcp_port; - use tide_disco::{app::AppHealth, healthcheck::HealthStatus}; use tokio::time::sleep; use vbs::version::StaticVersion; use versions::{ diff --git a/crates/espresso/node/src/api/data_source.rs b/crates/espresso/node/src/api/data_source.rs index f2ab7cfa790..68a3ea86e6a 100644 --- a/crates/espresso/node/src/api/data_source.rs +++ b/crates/espresso/node/src/api/data_source.rs @@ -35,7 +35,7 @@ use hotshot_types::{ use indexmap::IndexMap; use light_client::{state::LightClientOptions, storage::LightClientSqliteOptions}; use serde::{Deserialize, Serialize}; -use tide_disco::Url; +use url::Url; use super::{ AccountQueryData, BlocksFrontier, fs, diff --git a/crates/espresso/node/src/api/endpoints.rs b/crates/espresso/node/src/api/endpoints.rs deleted file mode 100644 index 55ecbba3b2d..00000000000 --- a/crates/espresso/node/src/api/endpoints.rs +++ /dev/null @@ -1,1161 +0,0 @@ -//! Sequencer-specific API endpoint handlers. - -use std::{ - collections::{BTreeSet, HashMap}, - env, -}; - -use alloy::primitives::utils::format_ether; -use anyhow::{Context, Result}; -use committable::Committable; -use espresso_types::{ - FeeAccount, FeeMerkleTree, PubKey, Transaction, - v0_3::RewardAccountV1, - v0_4::{RewardAccountV2, RewardClaimError}, -}; - -use crate::api::{ - RewardAmount, RewardMerkleTreeV2Data, data_source::TokenDataSource, unlock_schedule, -}; -// re-exported here to avoid breaking changes in consumers -// "deprecated" does not work with "pub use": https://github.com/rust-lang/rust/issues/30827 -#[deprecated(note = "use espresso_types::ADVZNamespaceProofQueryData")] -pub type ADVZNamespaceProofQueryData = espresso_types::ADVZNamespaceProofQueryData; -#[deprecated(note = "use espresso_types::NamespaceProofQueryData")] -pub type NamespaceProofQueryData = espresso_types::NamespaceProofQueryData; - -use futures::FutureExt; -use hotshot_query_service::{ - Error, - availability::AvailabilityDataSource, - explorer::{self, ExplorerDataSource}, - merklized_state::{ - self, MerklizedState, MerklizedStateDataSource, MerklizedStateHeightPersistence, Snapshot, - }, - node::{self, NodeDataSource}, -}; -use hotshot_types::{ - data::{EpochNumber, ViewNumber}, - traits::network::ConnectedNetwork, -}; -use jf_merkle_tree_compat::MerkleTreeScheme; -use serde::de::Error as _; -use tagged_base64::TaggedBase64; -use tide_disco::{Api, Error as _, RequestParams, StatusCode, method::ReadState}; -use vbs::version::{StaticVersion, StaticVersionType}; - -use super::data_source::{ - CatchupDataSource, DatabaseMetadataSource, HotShotConfigDataSource, NodeStateDataSource, - PruningDataSource, StakeTableDataSource, StateSignatureDataSource, SubmitDataSource, -}; -use crate::{ - SeqTypes, SequencerApiVersion, SequencerPersistence, api::RewardMerkleTreeDataSource, - options::PublicNodeConfig, -}; - -pub(crate) mod availability; -pub(super) use availability::*; - -pub(super) fn fee( - api_ver: semver::Version, -) -> Result> -where - State: 'static + Send + Sync + ReadState, - Ver: 'static + StaticVersionType, - ::State: Send - + Sync - + MerklizedStateDataSource - + MerklizedStateHeightPersistence, -{ - let mut options = merklized_state::Options::default(); - let extension = toml::from_str(include_str!("../../api/fee.toml"))?; - options.extensions.push(extension); - - let mut api = - merklized_state::define_api::(&options, api_ver)?; - - api.get("getfeebalance", move |req, state| { - async move { - let address = req.string_param("address")?; - let height = state.get_last_state_height().await?; - let snapshot = Snapshot::Index(height as u64); - let key = address - .parse() - .map_err(|_| merklized_state::Error::Custom { - message: "failed to parse address".to_string(), - status: StatusCode::BAD_REQUEST, - })?; - let path = state.get_path(snapshot, key).await?; - Ok(path.elem().copied()) - } - .boxed() - })?; - Ok(api) -} - -pub enum RewardMerkleTreeVersion { - V1, - V2, -} - -pub(super) fn reward( - api_ver: semver::Version, - merkle_tree_version: RewardMerkleTreeVersion, -) -> Result> -where - State: 'static + Send + Sync + ReadState, - Ver: 'static + StaticVersionType, - MT: MerklizedState, - for<'a> >::Error: std::fmt::Display, - >::Entry: std::marker::Copy, - ::State: Send - + Sync - + RewardMerkleTreeDataSource - + MerklizedStateDataSource - + MerklizedStateHeightPersistence, -{ - let mut options = merklized_state::Options::default(); - let extension = toml::from_str(include_str!("../../api/reward.toml"))?; - options.extensions.push(extension); - - let mut api = - merklized_state::define_api::(&options, api_ver)?; - - api.get("get_latest_reward_balance", move |req, state| { - async move { - let address = req.string_param("address")?; - let account = address - .parse() - .map_err(|_| merklized_state::Error::Custom { - message: format!("invalid reward address: {address}"), - status: StatusCode::BAD_REQUEST, - })?; - - state - .load_latest_reward_account_proof_v2(account) - .await - .map_err(|err| merklized_state::Error::Custom { - message: format!( - "failed to load latest reward account proof from storage for account \ - {account}: {err}" - ), - status: StatusCode::NOT_FOUND, - }) - .map(|proof| RewardAmount(proof.balance)) - } - .boxed() - })? - .get("get_latest_reward_account_proof", move |req, state| { - async move { - let address = req.string_param("address")?; - let account = address - .parse() - .map_err(|_| merklized_state::Error::Custom { - message: format!("invalid reward address: {address}"), - status: StatusCode::BAD_REQUEST, - })?; - - state - .load_latest_reward_account_proof_v2(account) - .await - .map_err(|err| merklized_state::Error::Custom { - message: format!( - "failed to load latest reward account proof from storage for account \ - {account}: {err}" - ), - status: StatusCode::NOT_FOUND, - }) - } - .boxed() - })? - .get("get_reward_balance", move |req, state| { - async move { - let address = req.string_param("address")?; - let height = req.integer_param("height")?; - let account = address - .parse() - .map_err(|_| merklized_state::Error::Custom { - message: format!("invalid reward address: {address}"), - status: StatusCode::BAD_REQUEST, - })?; - - state - .load_reward_account_proof_v2(height, account) - .await - .map_err(|err| merklized_state::Error::Custom { - message: format!( - "failed to load v2 reward account {address} at height {height}: {err}" - ), - status: StatusCode::NOT_FOUND, - }) - .map(|proof| Some(RewardAmount(proof.balance))) - } - .boxed() - })? - .get("get_reward_amounts", move |req, state| { - async move { - let height = req.integer_param("height")?; - let limit: usize = req.integer_param("limit")?; - let offset = req.integer_param("offset")?; - - if limit > 10_000 { - return Err(merklized_state::Error::Custom { - message: format!("limit {limit} exceeds maximum allowed 10000"), - status: StatusCode::BAD_REQUEST, - }); - } - - let tree_bytes = - state - .load_tree(height) - .await - .map_err(|err| merklized_state::Error::Custom { - message: format!( - "failed to load RewardMerkleTreeV2Data from storage at height \ - {height}: {err}" - ), - status: StatusCode::NOT_FOUND, - })?; - - let tree_data = bincode::deserialize::(&tree_bytes) - .context( - "Failed to deserialize RewardMerkleTreeV2 for height {height} from storage; \ - this should never happen.", - ) - .map_err(|err| merklized_state::Error::Custom { - message: format!( - "failed to load RewardMerkleTreeV2Data from storage at height {height}: \ - {err}" - ), - status: StatusCode::NOT_FOUND, - })?; - - let end = std::cmp::min(offset + limit, tree_data.balances.len()); - - let result = tree_data - .balances - .get(offset..end) - .ok_or(merklized_state::Error::Custom { - message: "Range out of bounds for balances".to_string(), - status: StatusCode::NOT_FOUND, - })? - .iter() - .rev() - .cloned() - .collect::>(); - - Ok(result) - } - .boxed() - })? - .get("get_reward_merkle_tree_v2", move |req, state| { - async move { - let height = req.integer_param("height")?; - - state - .load_tree(height) - .await - .map_err(|err| merklized_state::Error::Custom { - message: format!( - "failed to load RewardMerkleTreeV2Data from storage at height {height}: \ - {err}" - ), - status: StatusCode::NOT_FOUND, - }) - } - .boxed() - })?; - - match merkle_tree_version { - RewardMerkleTreeVersion::V1 => { - api.get("get_reward_account_proof", move |req, state| { - async move { - let address = req.string_param("address")?; - let height = req.integer_param("height")?; - let account = address - .parse() - .map_err(|_| merklized_state::Error::Custom { - message: format!("invalid reward address: {address}"), - status: StatusCode::BAD_REQUEST, - })?; - - state - .load_v1_reward_account_proof(height, account) - .await - .map_err(|err| merklized_state::Error::Custom { - message: format!( - "failed to load v1 reward account {address} at height {height}: \ - {err}" - ), - status: StatusCode::NOT_FOUND, - }) - } - .boxed() - })?; - }, - RewardMerkleTreeVersion::V2 => { - api.get("get_reward_account_proof", move |req, state| { - async move { - let address = req.string_param("address")?; - let height = req.integer_param("height")?; - let account = address - .parse() - .map_err(|_| merklized_state::Error::Custom { - message: format!("invalid reward address: {address}"), - status: StatusCode::BAD_REQUEST, - })?; - - state - .load_reward_account_proof_v2(height, account) - .await - .map_err(|err| merklized_state::Error::Custom { - message: format!( - "failed to load v2 reward account {address} at height {height}: \ - {err}" - ), - status: StatusCode::NOT_FOUND, - }) - } - .boxed() - })?; - - api.get("get_reward_claim_input", move |req, state| { - async move { - let address = req.string_param("address")?; - let height = req.integer_param("height")?; - let account = address - .parse() - .map_err(|_| merklized_state::Error::Custom { - message: format!("invalid reward address: {address}"), - status: StatusCode::BAD_REQUEST, - })?; - - let proof = state - .load_reward_account_proof_v2(height, account) - .await - .map_err(|err| merklized_state::Error::Custom { - message: format!( - "failed to load v2 reward account {address} at height {height}: \ - {err}" - ), - status: StatusCode::NOT_FOUND, - })?; - - // Auth root inputs (other than the reward merkle tree root) are currently - // all zero placeholder values. This may be extended in the future. - let claim_input = match proof.to_reward_claim_input() { - Ok(input) => input, - Err(RewardClaimError::ZeroRewardError) => { - return Err(merklized_state::Error::Custom { - message: format!( - "zero reward balance for {address} at height {height}" - ), - status: StatusCode::NOT_FOUND, - }); - }, - Err(RewardClaimError::ProofConversionError(err)) => { - let message = format!( - "failed to create solidity proof for {address} at height \ - {height}: {err}", - ); - tracing::warn!("{message}"); - // Normally we would not want to return the internal error via the - // API response but this is an error that should never occur. No - // secret data involved so it seems fine to return it. - return Err(merklized_state::Error::Custom { - message, - status: StatusCode::INTERNAL_SERVER_ERROR, - }); - }, - }; - - Ok(claim_input) - } - .boxed() - })?; - }, - } - - Ok(api) -} - -type ExplorerApi = Api, explorer::Error, ApiVer>; - -pub(super) fn explorer( - api_ver: semver::Version, -) -> Result> -where - N: ConnectedNetwork, - D: ExplorerDataSource + Send + Sync + 'static, - P: SequencerPersistence, -{ - let api = explorer::define_api::, SeqTypes, _>( - SequencerApiVersion::instance(), - api_ver, - )?; - Ok(api) -} - -pub(super) fn token(api_ver: semver::Version) -> Result>> -where - S: 'static + Send + Sync + ReadState, - ::State: Send - + Sync - + TokenDataSource - + NodeDataSource - + NodeStateDataSource - + AvailabilityDataSource, -{ - // Extend the base API - let mut options = node::Options::default(); - let extension = toml::from_str(include_str!("../../api/token.toml"))?; - options.extensions.push(extension); - - // Create the base API with our extensions - let mut api = - node::define_api::(&options, SequencerApiVersion::instance(), api_ver)?; - - // Tack on the application logic - api.at("get_total_minted_supply", |_, state| { - async move { - let value = state - .read(|state| state.get_total_supply_l1().boxed()) - .await - .map_err(|err| node::Error::Custom { - message: format!("failed to get total supply. err={err:#}"), - status: StatusCode::NOT_FOUND, - })?; - - Ok(format_ether(value)) - } - .boxed() - })?; - - api.at("get_circulating_supply", |_, state| { - async move { - let calc = fetch_supply_inputs(state).await?; - Ok(format_ether(calc.circulating_supply())) - } - .boxed() - })?; - - api.at("get_circulating_supply_ethereum", |_, state| { - async move { - let calc = fetch_supply_inputs(state).await?; - Ok(format_ether(calc.circulating_supply_ethereum())) - } - .boxed() - })?; - - api.at("get_total_issued_supply", |_, state| { - async move { - let calc = fetch_supply_inputs(state).await?; - Ok(format_ether(calc.total_issued_supply())) - } - .boxed() - })?; - - // Reuses fetch_supply_inputs for uniformity; the extra Ethereum fetches are cached. - api.at("get_total_reward_distributed", |_, state| { - async move { - let calc = fetch_supply_inputs(state).await?; - Ok(format_ether(calc.total_reward_distributed())) - } - .boxed() - })?; - - Ok(api) -} - -/// Fetch state data and build a [`unlock_schedule::SupplyCalculator`]. -async fn fetch_supply_inputs( - state: &S, -) -> Result -where - S::State: Send + Sync + TokenDataSource + NodeStateDataSource, -{ - let node_state = state.read(|s| s.node_state().boxed()).await; - let chain_id = node_state.chain_config.chain_id; - - let header = state.read(|s| s.get_decided_header().boxed()).await; - let now_secs = header.timestamp_internal(); - let total_reward_distributed = header.total_reward_distributed(); - - let initial_supply = state - .read(|s| s.get_initial_supply_l1().boxed()) - .await - .map_err(|err| node::Error::Custom { - message: format!("failed to get initial supply: {err:#}"), - status: StatusCode::INTERNAL_SERVER_ERROR, - })?; - - let total_supply_l1 = state - .read(|s| s.get_total_supply_l1().boxed()) - .await - .map_err(|err| node::Error::Custom { - message: format!("failed to get total supply: {err:#}"), - status: StatusCode::INTERNAL_SERVER_ERROR, - })?; - - Ok(unlock_schedule::SupplyCalculator::new( - chain_id, - now_secs, - initial_supply, - total_supply_l1, - total_reward_distributed, - )) -} - -pub(super) fn node(api_ver: semver::Version) -> Result>> -where - S: 'static + Send + Sync + ReadState, - ::State: Send - + Sync - + StakeTableDataSource - + NodeDataSource - + AvailabilityDataSource - + PruningDataSource, -{ - // Extend the base API - let mut options = node::Options::default(); - let extension = toml::from_str(include_str!("../../api/node.toml"))?; - options.extensions.push(extension); - - // Create the base API with our extensions - let mut api = - node::define_api::(&options, SequencerApiVersion::instance(), api_ver)?; - - // Tack on the application logic - api.at("stake_table", |req, state| { - async move { - // Try to get the epoch from the request. If this fails, error - // as it was probably a mistake - let epoch = req - .opt_integer_param("epoch_number") - .map_err(|_| hotshot_query_service::node::Error::Custom { - message: "Epoch number is required".to_string(), - status: StatusCode::BAD_REQUEST, - })? - .map(EpochNumber::new); - - state - .read(|state| state.get_stake_table(epoch).boxed()) - .await - .map_err(|err| node::Error::Custom { - message: format!("failed to get stake table for epoch={epoch:?}. err={err:#}"), - status: StatusCode::NOT_FOUND, - }) - } - .boxed() - })? - .at("stake_table_current", |_, state| { - async move { - state - .read(|state| state.get_stake_table_current().boxed()) - .await - .map_err(|err| node::Error::Custom { - message: format!("failed to get current stake table. err={err:#}"), - status: StatusCode::NOT_FOUND, - }) - } - .boxed() - })? - .at("da_stake_table", |req, state| { - async move { - // Try to get the epoch from the request. If this fails, error - // as it was probably a mistake - let epoch = req - .opt_integer_param("epoch_number") - .map_err(|_| hotshot_query_service::node::Error::Custom { - message: "Epoch number is required".to_string(), - status: StatusCode::BAD_REQUEST, - })? - .map(EpochNumber::new); - - state - .read(|state| state.get_da_stake_table(epoch).boxed()) - .await - .map_err(|err| node::Error::Custom { - message: format!( - "failed to get DA stake table for epoch={epoch:?}. err={err:#}" - ), - status: StatusCode::NOT_FOUND, - }) - } - .boxed() - })? - .at("da_stake_table_current", |_, state| { - async move { - state - .read(|state| state.get_da_stake_table_current().boxed()) - .await - .map_err(|err| node::Error::Custom { - message: format!("failed to get current DA stake table. err={err:#}"), - status: StatusCode::NOT_FOUND, - }) - } - .boxed() - })? - .at("get_validators", |req, state| { - async move { - let epoch = req.integer_param::<_, u64>("epoch_number").map_err(|_| { - hotshot_query_service::node::Error::Custom { - message: "Epoch number is required".to_string(), - status: StatusCode::BAD_REQUEST, - } - })?; - - state - .read(|state| state.get_validators(EpochNumber::new(epoch)).boxed()) - .await - .map_err(|err| hotshot_query_service::node::Error::Custom { - message: format!("failed to get validators mapping: err: {err}"), - status: StatusCode::NOT_FOUND, - }) - } - .boxed() - })? - .at("get_all_validators", |req, state| { - async move { - let epoch = req.integer_param::<_, u64>("epoch_number").map_err(|_| { - hotshot_query_service::node::Error::Custom { - message: "Epoch number is required".to_string(), - status: StatusCode::BAD_REQUEST, - } - })?; - - let offset = req.integer_param::<_, u64>("offset")?; - - let limit = req.integer_param::<_, u64>("limit")?; - if limit > 1000 { - return Err(hotshot_query_service::node::Error::Custom { - message: "Limit cannot be greater than 1000".to_string(), - status: StatusCode::BAD_REQUEST, - }); - } - - state - .read(|state| { - state - .get_all_validators(EpochNumber::new(epoch), offset, limit) - .boxed() - }) - .await - .map_err(|err| hotshot_query_service::node::Error::Custom { - message: format!("failed to get all validators : err: {err}"), - status: StatusCode::INTERNAL_SERVER_ERROR, - }) - } - .boxed() - })? - .at("current_proposal_participation", |_, state| { - async move { - Ok(state - .read(|state| state.current_proposal_participation().boxed()) - .await) - } - .boxed() - })? - .at("proposal_participation", |req, state| { - async move { - let epoch = req.integer_param::<_, u64>("epoch").map_err(|_| { - hotshot_query_service::node::Error::Custom { - message: "Epoch number is required".to_string(), - status: StatusCode::BAD_REQUEST, - } - })?; - - Ok(state - .read(|state| state.proposal_participation(epoch.into()).boxed()) - .await) - } - .boxed() - })? - .at("current_vote_participation", |_, state| { - async move { - Ok(state - .read(|state| state.current_vote_participation().boxed()) - .await) - } - .boxed() - })? - .at("vote_participation", |req, state| { - async move { - let epoch = req.integer_param::<_, u64>("epoch").map_err(|_| { - hotshot_query_service::node::Error::Custom { - message: "Epoch number is required".to_string(), - status: StatusCode::BAD_REQUEST, - } - })?; - - Ok(state - .read(|state| state.vote_participation(epoch.into()).boxed()) - .await) - } - .boxed() - })? - .at("get_block_reward", |req, state| { - async move { - let epoch = req - .opt_integer_param::<_, u64>("epoch_number")? - .map(EpochNumber::new); - - state - .read(|state| state.get_block_reward(epoch).boxed()) - .await - .map_err(|err| node::Error::Custom { - message: format!("failed to get block reward. err={err:#}"), - status: StatusCode::NOT_FOUND, - }) - } - .boxed() - })? - .at("get_oldest_block", |_req, state| { - async move { - state - .read(|state| state.get_oldest_block().boxed()) - .await - .map_err(|err| node::Error::Custom { - message: format!("failed to get oldest block: {err:#}"), - status: StatusCode::INTERNAL_SERVER_ERROR, - }) - } - .boxed() - })? - .at("get_oldest_leaf", |_req, state| { - async move { - state - .read(|state| state.get_oldest_leaf().boxed()) - .await - .map_err(|err| node::Error::Custom { - message: format!("failed to get oldest leaf: {err:#}"), - status: StatusCode::INTERNAL_SERVER_ERROR, - }) - } - .boxed() - })?; - - Ok(api) -} - -pub(super) fn database( - api_ver: semver::Version, -) -> Result> -where - S: 'static + Send + Sync + ReadState, - ::State: Send + Sync + DatabaseMetadataSource, -{ - let toml = toml::from_str::(include_str!("../../api/database.toml"))?; - let mut api = Api::::new(toml)?; - - api.with_version(api_ver) - .at("get_table_sizes", |_req, state| { - async move { - state - .read(|state| state.get_table_sizes().boxed()) - .await - .map_err(|err| Error::internal(format!("failed to get table sizes: {err:#}"))) - } - .boxed() - })? - .at("get_migration_status", |_req, state| { - async move { - state - .read(|state| state.get_migration_status().boxed()) - .await - .map_err(|err| { - Error::internal(format!("failed to get migration status: {err:#}")) - }) - } - .boxed() - })?; - - Ok(api) -} - -pub(super) fn submit( - api_ver: semver::Version, -) -> Result> -where - N: ConnectedNetwork, - S: 'static + Send + Sync + ReadState, - P: SequencerPersistence, - S::State: Send + Sync + SubmitDataSource, -{ - let toml = toml::from_str::(include_str!("../../api/submit.toml"))?; - let mut api = Api::::new(toml)?; - - api.with_version(api_ver).at("submit", |req, state| { - async move { - let tx = req - .body_auto::(ApiVer::instance()) - .map_err(Error::from_request_error)?; - - let hash = tx.commit(); - state - .read(|state| state.submit(tx).boxed()) - .await - .map_err(|err| Error::internal(err.to_string()))?; - Ok(hash) - } - .boxed() - })?; - - Ok(api) -} - -pub(super) fn state_signature( - _: ApiVer, - api_ver: semver::Version, -) -> Result> -where - N: ConnectedNetwork, - S: 'static + Send + Sync + ReadState, - S::State: Send + Sync + StateSignatureDataSource, -{ - let toml = toml::from_str::(include_str!("../../api/state_signature.toml"))?; - let mut api = Api::::new(toml)?; - api.with_version(api_ver); - - api.get("get_state_signature", |req, state| { - async move { - let height = req - .integer_param("height") - .map_err(Error::from_request_error)?; - state - .get_state_signature(height) - .await - .ok_or(tide_disco::Error::catch_all( - StatusCode::NOT_FOUND, - "Signature not found.".to_owned(), - )) - } - .boxed() - })?; - - Ok(api) -} - -pub(super) fn catchup( - _: ApiVer, - api_ver: semver::Version, -) -> Result> -where - S: 'static + Send + Sync + ReadState, - S::State: Send + Sync + NodeStateDataSource + CatchupDataSource, -{ - let toml = toml::from_str::(include_str!("../../api/catchup.toml"))?; - let mut api = Api::::new(toml)?; - api.with_version(api_ver); - - let parse_height_view = |req: &RequestParams| -> Result<(u64, ViewNumber), Error> { - let height = req - .integer_param("height") - .map_err(Error::from_request_error)?; - let view = req - .integer_param("view") - .map_err(Error::from_request_error)?; - Ok((height, ViewNumber::new(view))) - }; - - let parse_fee_account = |req: &RequestParams| -> Result { - let raw = req - .string_param("address") - .map_err(Error::from_request_error)?; - raw.parse().map_err(|err| { - Error::catch_all( - StatusCode::BAD_REQUEST, - format!("malformed fee account {raw}: {err}"), - ) - }) - }; - - let parse_reward_account = |req: &RequestParams| -> Result { - let raw = req - .string_param("address") - .map_err(Error::from_request_error)?; - raw.parse().map_err(|err| { - Error::catch_all( - StatusCode::BAD_REQUEST, - format!("malformed reward account {raw}: {err}"), - ) - }) - }; - - api.get("account", move |req, state| { - async move { - let (height, view) = parse_height_view(&req)?; - let account = parse_fee_account(&req)?; - state - .get_account(&state.node_state().await, height, view, account) - .await - .map_err(|err| Error::catch_all(StatusCode::NOT_FOUND, format!("{err:#}"))) - } - .boxed() - })? - .at("accounts", move |req, state| { - async move { - let (height, view) = parse_height_view(&req)?; - let accounts = req - .body_auto::, ApiVer>(ApiVer::instance()) - .map_err(Error::from_request_error)?; - - state - .read(|state| { - async move { - state - .get_accounts(&state.node_state().await, height, view, &accounts) - .await - .map_err(|err| { - Error::catch_all(StatusCode::NOT_FOUND, format!("{err:#}")) - }) - } - .boxed() - }) - .await - } - .boxed() - })? - .get("reward_account", move |req, state| { - async move { - let (height, view) = parse_height_view(&req)?; - let account = parse_reward_account(&req)?; - state - .get_reward_account_v1(&state.node_state().await, height, view, account.into()) - .await - .map_err(|err| Error::catch_all(StatusCode::NOT_FOUND, format!("{err:#}"))) - } - .boxed() - })? - .at("reward_accounts", move |req, state| { - async move { - let (height, view) = parse_height_view(&req)?; - let accounts = req - .body_auto::, ApiVer>(ApiVer::instance()) - .map_err(Error::from_request_error)?; - - state - .read(|state| { - async move { - state - .get_reward_accounts_v1( - &state.node_state().await, - height, - view, - &accounts, - ) - .await - .map_err(|err| { - Error::catch_all(StatusCode::NOT_FOUND, format!("{err:#}")) - }) - } - .boxed() - }) - .await - } - .boxed() - })? - .get("reward_account_v2", move |req, state| { - async move { - let (height, view) = parse_height_view(&req)?; - let account = parse_reward_account(&req)?; - - state - .get_reward_account_v2(&state.node_state().await, height, view, account) - .await - .map_err(|err| Error::catch_all(StatusCode::NOT_FOUND, format!("{err:#}"))) - } - .boxed() - })? - .get("reward_amounts", move |_req, _state| { - async move { - Err::(Error::catch_all( - StatusCode::NOT_FOUND, - "catchup/reward-amounts is deprecated".to_string(), - )) - } - .boxed() - })? - .at("reward_accounts_v2", move |_req, _state| { - async move { - Err::(Error::catch_all( - StatusCode::NOT_FOUND, - "catchup/reward-accounts-v2 is deprecated".to_string(), - )) - } - .boxed() - })? - .get("blocks", |req, state| { - async move { - let height = req - .integer_param("height") - .map_err(Error::from_request_error)?; - let view = req - .integer_param("view") - .map_err(Error::from_request_error)?; - - state - .get_frontier(&state.node_state().await, height, ViewNumber::new(view)) - .await - .map_err(|err| Error::catch_all(StatusCode::NOT_FOUND, format!("{err:#}"))) - } - .boxed() - })? - .get("chainconfig", |req, state| { - async move { - let commitment = req - .blob_param("commitment") - .map_err(Error::from_request_error)?; - - state - .get_chain_config(commitment) - .await - .map_err(|err| Error::catch_all(StatusCode::NOT_FOUND, format!("{err:#}"))) - } - .boxed() - })? - .get("leafchain", |req, state| { - async move { - let height = req - .integer_param("height") - .map_err(Error::from_request_error)?; - state - .get_leaf_chain(height) - .await - .map_err(|err| Error::catch_all(StatusCode::NOT_FOUND, format!("{err:#}"))) - } - .boxed() - })? - .get("cert2", |req, state| { - async move { - let height = req - .integer_param("height") - .map_err(Error::from_request_error)?; - let response = state - .get_cert2(height) - .await - .map_err(|err| Error::catch_all(StatusCode::NOT_FOUND, format!("{err:#}")))?; - response.ok_or_else(|| { - Error::catch_all( - StatusCode::NOT_FOUND, - format!("no cert2 available for height {height}"), - ) - }) - } - .boxed() - })? - .get("reward_merkle_tree_v2", move |req, state| { - async move { - let (height, view) = parse_height_view(&req)?; - state - .get_reward_merkle_tree_v2(height, view) - .await - .map_err(|err| Error::catch_all(StatusCode::NOT_FOUND, format!("{err:#}"))) - } - .boxed() - })? - .get("state_cert", |req, state| { - async move { - let epoch = req - .integer_param("epoch") - .map_err(Error::from_request_error)?; - state - .get_state_cert(epoch) - .await - .map_err(|err| Error::catch_all(StatusCode::NOT_FOUND, format!("{err:#}"))) - } - .boxed() - })?; - - Ok(api) -} - -type MerklizedStateApi = Api, merklized_state::Error, ApiVer>; -pub(super) fn merklized_state( - api_ver: semver::Version, -) -> Result> -where - N: ConnectedNetwork, - D: MerklizedStateDataSource - + Send - + Sync - + MerklizedStateHeightPersistence - + 'static, - S: MerklizedState, - P: SequencerPersistence, - for<'a> >::Error: std::fmt::Display, -{ - let api = merklized_state::define_api::< - AvailState, - SeqTypes, - S, - SequencerApiVersion, - ARITY, - >(&Default::default(), api_ver)?; - Ok(api) -} - -pub(super) fn config( - _: ApiVer, - api_ver: semver::Version, - public_node_config: Option, -) -> Result> -where - S: 'static + Send + Sync + ReadState, - S::State: Send + Sync + HotShotConfigDataSource, -{ - let toml = toml::from_str::(include_str!("../../api/config.toml"))?; - let mut api = Api::::new(toml)?; - api.with_version(api_ver); - - let env_variables = get_public_env_vars() - .map_err(|err| Error::catch_all(StatusCode::INTERNAL_SERVER_ERROR, format!("{err:#}")))?; - - api.get("hotshot", |_, state| { - async move { Ok(state.get_config().await) }.boxed() - })? - .get("env", move |_, _| { - { - let env_variables = env_variables.clone(); - async move { Ok(env_variables) } - } - .boxed() - })? - .get("runtime", move |_, _| { - let public_node_config = public_node_config.clone(); - async move { - public_node_config.ok_or_else(|| { - Error::catch_all( - StatusCode::NOT_FOUND, - "runtime config not available".to_string(), - ) - }) - } - .boxed() - })?; - - Ok(api) -} - -pub(super) fn get_public_env_vars() -> Result> { - let toml: toml::Value = toml::from_str(include_str!("../../api/public-env-vars.toml"))?; - - let keys = toml - .get("variables") - .ok_or_else(|| toml::de::Error::custom("variables not found"))? - .as_array() - .ok_or_else(|| toml::de::Error::custom("variables is not an array"))? - .clone() - .into_iter() - .map(|v| v.try_into()) - .collect::, toml::de::Error>>()?; - - let hashmap: HashMap = env::vars().collect(); - let mut public_env_vars: Vec = Vec::new(); - for key in keys { - let value = hashmap.get(&key).cloned().unwrap_or_default(); - public_env_vars.push(format!("{key}={value}")); - } - - Ok(public_env_vars) -} diff --git a/crates/espresso/node/src/api/endpoints/availability.rs b/crates/espresso/node/src/api/endpoints/availability.rs deleted file mode 100644 index b079fccec9a..00000000000 --- a/crates/espresso/node/src/api/endpoints/availability.rs +++ /dev/null @@ -1,452 +0,0 @@ -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 = ApiState>; - -type AvailabilityApi = Api, 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( - block: &BlockQueryData, - common: &VidCommonQueryData, - ns_id: NamespaceId, - state: &S, -) -> Result, Error> -where - S: ReadState, - S::State: NodeDataSource + RequestResponseDataSource + 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 = 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, - ns_id: NamespaceId, -) -> Result { - 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, - ns_id: NamespaceId, -) -> Result { - 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( - req: &RequestParams, - state: &S, - timeout: Duration, -) -> Result<(BlockQueryData, VidCommonQueryData), Error> -where - S: ReadState, - S::State: AvailabilityDataSource + 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( - req: &RequestParams, - state: &S, - limit: usize, - timeout: Duration, -) -> Result, VidCommonQueryData)>, Error> -where - S: ReadState, - S::State: AvailabilityDataSource + 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, - VidCommonQueryData, - ), - Error, - >, -> -where - S: ReadState, - S::State: AvailabilityDataSource + 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( - state: &S, - epoch: u64, - timeout: Duration, -) -> Result, availability::Error> -where - S: ReadState, - S::State: StateCertDataSource + StateCertFetchingDataSource + 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( - api_ver: semver::Version, -) -> anyhow::Result> -where - N: ConnectedNetwork, - 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::, 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) -} diff --git a/crates/espresso/node/src/api/light_client.rs b/crates/espresso/node/src/api/light_client.rs index 7fcca98f596..e27da9103c7 100644 --- a/crates/espresso/node/src/api/light_client.rs +++ b/crates/espresso/node/src/api/light_client.rs @@ -2,37 +2,24 @@ 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::{FutureExt, join, try_join}, - stream::StreamExt, -}; +use futures::{TryStreamExt, future::try_join, stream::StreamExt}; use hotshot_query_service::{ Error, - availability::{ - self, AvailabilityDataSource, LeafId, LeafQueryData, PayloadQueryData, VidCommonQueryData, - }, + availability::{AvailabilityDataSource, LeafQueryData, PayloadQueryData, VidCommonQueryData}, data_source::{VersionedDataSource, storage::NodeStorage}, merklized_state::{MerklizedStateDataSource, Snapshot}, node::BlockId, types::HeightIndexed, }; -use hotshot_types::utils::{epoch_from_block_number, root_block_in_epoch}; use itertools::izip; use jf_merkle_tree_compat::MerkleTreeScheme; use light_client::{ client::NAMESPACES_PARAM_TAG, - consensus::{ - header::HeaderProof, leaf::LeafProof, namespace::NamespaceProof, payload::PayloadProof, - }, + consensus::{header::HeaderProof, leaf::LeafProof, namespace::NamespaceProof}, }; use tagged_base64::TaggedBase64; -use tide_disco::{Api, RequestParams, StatusCode, method::ReadState}; -use vbs::version::StaticVersionType; -use versions::NEW_PROTOCOL_VERSION; - -use crate::api::data_source::{NodeStateDataSource, StakeTableDataSource}; pub(crate) async fn get_leaf_proof_with_qc_chain( state: &State, @@ -398,16 +385,6 @@ pub(crate) fn parse_namespaces_str(encoded: &str) -> anyhow::Result> { .map_err(|err| anyhow::anyhow!("invalid namespaces parameter: {err}")) } -fn parse_namespaces_param(req: &RequestParams) -> Result, Error> { - let encoded = req - .tagged_base64_param("namespaces") - .map_err(bad_param("namespaces"))?; - parse_namespaces_str(&encoded.to_string()).map_err(|err| Error::Custom { - message: err.to_string(), - status: StatusCode::BAD_REQUEST, - }) -} - /// Construct a [`NamespaceProof`] for the namespace at `ns_index` of the given block. fn build_namespace_proof( payload: &PayloadQueryData, @@ -424,396 +401,6 @@ fn build_namespace_proof( Ok(NamespaceProof::new(ns_proof, vid_common.common().clone())) } -#[derive(Debug)] -pub(super) struct Options { - /// Timeout for failing requests due to missing data. - /// - /// If data needed to respond to a request is missing, it can (in some cases) be fetched from an - /// external provider. This parameter controls how long the request handler will wait for - /// missing data to be fetched before giving up and failing the request. - pub fetch_timeout: Duration, - - /// The maximum number of large objects which can be loaded in a single range query. - /// - /// Large objects include anything that _might_ contain a full payload or an object proportional - /// in size to a payload. Note that this limit applies to the entire class of objects: we do not - /// check the size of objects while loading to determine which limit to apply. If an object - /// belongs to a class which might contain a large payload, the large object limit always - /// applies. - pub large_object_range_limit: usize, -} - -impl Default for Options { - fn default() -> Self { - Self { - fetch_timeout: Duration::from_millis(500), - large_object_range_limit: availability::Options::default().large_object_range_limit, - } - } -} - -pub(super) fn define_api( - opt: Options, - api_ver: semver::Version, -) -> Result> -where - S: ReadState + Send + Sync + 'static, - S::State: AvailabilityDataSource - + MerklizedStateDataSource - + NodeStateDataSource - + StakeTableDataSource - + VersionedDataSource, - for<'a> ::ReadOnly<'a>: NodeStorage, -{ - let toml = toml::from_str::(include_str!("../../api/light-client.toml"))?; - let mut api = Api::::new(toml)?; - api.with_version(api_ver); - - let Options { - fetch_timeout, - large_object_range_limit, - } = opt; - - api.get("leaf", move |req, state| { - async move { - let requested_leaf = leaf_from_req(&req, state, fetch_timeout).await?; - let finalized = req - .opt_integer_param("finalized") - .map_err(bad_param("finalized"))?; - - if let Some(finalized) = finalized { - get_leaf_proof_with_finalized_assumption( - state, - requested_leaf, - finalized, - fetch_timeout, - ) - .await - } else if requested_leaf.header().version() >= NEW_PROTOCOL_VERSION { - get_leaf_proof_with_cert2(state, requested_leaf, fetch_timeout).await - } else { - get_leaf_proof_with_qc_chain(state, requested_leaf, fetch_timeout).await - } - } - .boxed() - })? - .get("header", move |req, state| { - async move { - let root = req.integer_param("root").map_err(bad_param("root"))?; - let requested = block_id_from_req(&req)?; - get_header_proof(state, root, requested, fetch_timeout).await - } - .boxed() - })? - .get("stake_table", move |req, state| { - async move { - let epoch: u64 = req.integer_param("epoch").map_err(bad_param("epoch"))?; - - let node_state = state.node_state().await; - let epoch_height = node_state.epoch_height.ok_or_else(|| Error::Custom { - message: "epoch state not set".into(), - status: StatusCode::INTERNAL_SERVER_ERROR, - })?; - let first_epoch = epoch_from_block_number(node_state.epoch_start_block, epoch_height); - - if epoch < first_epoch + 2 { - return Err(Error::Custom { - message: format!("epoch must be at least {}", first_epoch + 2), - status: StatusCode::BAD_REQUEST, - }); - } - - // Find the range of L1 block containing events for this epoch. This is determined by - // the `l1_finalized` field of the epoch root (from two epochs prior) and the previous - // epoch's epoch root. - let epoch_root_height = root_block_in_epoch(epoch - 2, epoch_height) as usize; - let epoch_root = state - .get_header(epoch_root_height) - .await - .with_timeout(fetch_timeout) - .await - .ok_or_else(|| { - not_found(format!("missing epoch root header {epoch_root_height}")) - })?; - let to_l1_block = epoch_root - .l1_finalized() - .ok_or_else(|| Error::Custom { - message: "epoch root header is missing L1 finalized block".into(), - status: StatusCode::INTERNAL_SERVER_ERROR, - })? - .number(); - - let from_l1_block = if epoch >= first_epoch + 3 { - let prev_epoch_root_height = root_block_in_epoch(epoch - 3, epoch_height) as usize; - let prev_epoch_root = state - .get_header(prev_epoch_root_height) - .await - .with_timeout(fetch_timeout) - .await - .ok_or_else(|| { - not_found(format!( - "missing previous epoch root header {prev_epoch_root_height}" - )) - })?; - prev_epoch_root - .l1_finalized() - .ok_or_else(|| Error::Custom { - message: "previous epoch root header is missing L1 finalized block".into(), - status: StatusCode::INTERNAL_SERVER_ERROR, - })? - .number() - + 1 - } else { - 0 - }; - - state - .stake_table_events(from_l1_block, to_l1_block) - .await - .map_err(|err| Error::Custom { - message: format!("failed to load stake table events: {err:#}"), - status: StatusCode::INTERNAL_SERVER_ERROR, - }) - } - .boxed() - })? - .get("payload", move |req, state| { - async move { - let height: usize = req.integer_param("height").map_err(bad_param("height"))?; - let fetch_payload = async move { - state - .get_payload(height) - .await - .with_timeout(fetch_timeout) - .await - .ok_or_else(|| Error::Custom { - message: format!("missing payload {height}"), - status: StatusCode::NOT_FOUND, - }) - }; - let fetch_vid_common = async move { - state - .get_vid_common(height) - .await - .with_timeout(fetch_timeout) - .await - .ok_or_else(|| Error::Custom { - message: format!("missing VID common {height}"), - status: StatusCode::NOT_FOUND, - }) - }; - let (payload, vid_common) = try_join(fetch_payload, fetch_vid_common).await?; - Ok(PayloadProof::new( - payload.data().clone(), - vid_common.common().clone(), - )) - } - .boxed() - })? - .get("payload_range", move |req, state| { - async move { - let start: usize = req.integer_param("start").map_err(bad_param("start"))?; - let end: usize = req.integer_param("end").map_err(bad_param("end"))?; - let fetch_payloads = async move { - state.get_payload_range(start..end).await.enumerate().then( - move |(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, - }) - }, - ) - }; - let fetch_vid_commons = async move { - state - .get_vid_common_range(start..end) - .await - .enumerate() - .then(move |(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, - }) - }) - }; - let (payloads, vid_commons) = join(fetch_payloads, fetch_vid_commons).await; - payloads - .zip(vid_commons) - .map(|(payload, vid_common)| { - Ok(PayloadProof::new( - payload?.data().clone(), - vid_common?.common().clone(), - )) - }) - .try_collect::>() - .await - } - .boxed() - })? - .get("namespace", move |req, state| { - async move { - let height = req.integer_param("height").map_err(bad_param("height"))?; - let namespace = req - .integer_param("namespace") - .map_err(bad_param("namespace"))?; - let mut proofs = get_namespace_proof_range( - state, - height, - height + 1, - namespace, - fetch_timeout, - large_object_range_limit, - ) - .await?; - if proofs.len() != 1 { - tracing::error!( - height, - namespace, - ?proofs, - "get_namespace_proof_range should have returned exactly one proof" - ); - return Err(Error::Custom { - message: "internal consistency error".into(), - status: StatusCode::INTERNAL_SERVER_ERROR, - }); - } - Ok(proofs.remove(0)) - } - .boxed() - })? - .get("namespace_range", move |req, state| { - async move { - let start = req.integer_param("start").map_err(bad_param("start"))?; - let end = req.integer_param("end").map_err(bad_param("end"))?; - let namespace = req - .integer_param("namespace") - .map_err(bad_param("namespace"))?; - get_namespace_proof_range( - state, - start, - end, - namespace, - fetch_timeout, - large_object_range_limit, - ) - .await - } - .boxed() - })? - .get("namespaces_range", move |req, state| { - async move { - let start = req.integer_param("start").map_err(bad_param("start"))?; - let end = req.integer_param("end").map_err(bad_param("end"))?; - let namespaces = parse_namespaces_param(&req)?; - get_namespaces_proof_range( - state, - start, - end, - &namespaces, - fetch_timeout, - large_object_range_limit, - ) - .await - } - .boxed() - })?; - - Ok(api) -} - -async fn leaf_from_req( - req: &RequestParams, - state: &S, - fetch_timeout: Duration, -) -> Result, Error> -where - S: AvailabilityDataSource, -{ - let requested = if let Some(height) = req - .opt_integer_param::<_, usize>("height") - .map_err(bad_param("height"))? - { - LeafId::Number(height) - } else if let Some(hash) = req.opt_blob_param("hash").map_err(bad_param("hash"))? { - LeafId::Hash(hash) - } else if let Some(hash) = req - .opt_blob_param("block-hash") - .map_err(bad_param("block-hash"))? - { - let header = state - .get_header(BlockId::Hash(hash)) - .await - .with_timeout(fetch_timeout) - .await - .ok_or_else(|| not_found(format!("unknown block hash {hash}")))?; - LeafId::Number(header.height() as usize) - } else if let Some(hash) = req - .opt_blob_param("payload-hash") - .map_err(bad_param("payload-hash"))? - { - let header = state - .get_header(BlockId::PayloadHash(hash)) - .await - .with_timeout(fetch_timeout) - .await - .ok_or_else(|| not_found(format!("unknown payload hash {hash}")))?; - LeafId::Number(header.height() as usize) - } else { - return Err(Error::Custom { - message: "missing parameter: requested leaf must be identified by height, hash, block \ - hash, or payload hash" - .into(), - status: StatusCode::BAD_REQUEST, - }); - }; - - state - .get_leaf(requested) - .await - .with_timeout(fetch_timeout) - .await - .ok_or_else(|| not_found(format!("unknown leaf {requested}"))) -} - -fn block_id_from_req(req: &RequestParams) -> Result, Error> { - if let Some(height) = req - .opt_integer_param("height") - .map_err(bad_param("height"))? - { - Ok(BlockId::Number(height)) - } else if let Some(hash) = req.opt_blob_param("hash").map_err(bad_param("hash"))? { - Ok(BlockId::Hash(hash)) - } else if let Some(hash) = req - .opt_blob_param("payload-hash") - .map_err(bad_param("payload-hash"))? - { - Ok(BlockId::PayloadHash(hash)) - } else { - Err(Error::Custom { - message: "missing parameter: requested header must be identified by height, hash, or \ - payload hash" - .into(), - status: StatusCode::BAD_REQUEST, - }) - } -} - -fn bad_param(name: &'static str) -> impl FnOnce(E) -> Error -where - E: Display, -{ - move |err| Error::Custom { - message: format!("{name}: {err:#}"), - status: StatusCode::BAD_REQUEST, - } -} - fn internal(err: impl Display) -> Error { Error::Custom { message: err.to_string(), @@ -833,6 +420,7 @@ 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::{ @@ -849,7 +437,6 @@ mod test { leaf_chain, leaf_chain_with_upgrade, }, }; - use tide_disco::Error; use versions::{DRB_AND_HEADER_UPGRADE_VERSION, EPOCH_VERSION, NEW_PROTOCOL_VERSION}; use super::*; diff --git a/crates/espresso/node/src/api/options.rs b/crates/espresso/node/src/api/options.rs index 54db72caaae..cb83c157f79 100644 --- a/crates/espresso/node/src/api/options.rs +++ b/crates/espresso/node/src/api/options.rs @@ -1,49 +1,43 @@ //! Sequencer-specific API options and initialization. -use std::sync::Arc; +use std::{ + collections::{BTreeSet, HashMap}, + env, + sync::Arc, +}; use ::light_client::{state::LightClientOptions, storage::LightClientSqliteOptions}; use anyhow::{Context, bail}; use clap::Parser; use espresso_telemetry as telemetry; use espresso_types::{ - BlockMerkleTree, PubKey, SeqTypes, + PubKey, v0::traits::{EventConsumer, NullEventConsumer, PersistenceOptions, SequencerPersistence}, - v0_3::RewardMerkleTreeV1, - v0_4::RewardMerkleTreeV2, -}; -use futures::{ - channel::oneshot, - future::{BoxFuture, Future}, }; +use futures::{channel::oneshot, future::BoxFuture}; use hotshot_query_service::{ - ApiState as AppState, Error, data_source::{ExtensibleDataSource, MetricsDataSource}, - status::{self, HasMetrics, UpdateStatusData}, + status::{HasMetrics, UpdateStatusData}, }; use hotshot_types::traits::{ metrics::{Metrics, NoMetrics}, network::ConnectedNetwork, }; -use jf_merkle_tree_compat::MerkleTreeScheme; use process_metrics::ProcessMetrics; -use tide_disco::{Api, App, Url, listener::RateLimitListener, method::ReadState}; -use vbs::version::StaticVersionType; +use serde::de::Error as _; +use url::Url; use super::{ ApiState, StorageState, data_source::{ - CatchupDataSource, HotShotConfigDataSource, NodeStateDataSource, Provider, - PruningDataSource, SequencerDataSource, StateSignatureDataSource, SubmitDataSource, - provider, + NodeStateDataSource, Provider, PruningDataSource, SequencerDataSource, provider, }, - endpoints, fs, light_client, sql, + fs, sql, state::NodeApiStateImpl, update::ApiEventConsumer, }; use crate::{ - SequencerApiVersion, - api::{LightClientProvider, endpoints::RewardMerkleTreeVersion}, + api::LightClientProvider, catchup::CatchupStorage, context::{SequencerContext, TaskList}, options::PublicNodeConfig, @@ -197,23 +191,11 @@ impl Options { Option, ) = if let Some(query_opt) = self.query.take() { if let Some(opt) = self.storage_sql.take() { - self.init_with_query_module_sql( - query_opt, - opt, - state, - &mut tasks, - SequencerApiVersion::instance(), - ) - .await? + self.init_with_query_module_sql(query_opt, opt, state, &mut tasks) + .await? } else if let Some(opt) = self.storage_fs.take() { - self.init_with_query_module_fs( - query_opt, - opt, - state, - &mut tasks, - SequencerApiVersion::instance(), - ) - .await? + self.init_with_query_module_fs(query_opt, opt, state, &mut tasks) + .await? } else { bail!("query module requested but not storage provided"); } @@ -225,28 +207,10 @@ impl Options { let metrics = ds.populate_metrics(); telemetry::set_registry(Arc::new(ds.metrics().registry().clone())); tasks.spawn("process_metrics", ProcessMetrics::new(ds.metrics()).run()); - let axum_ds = Arc::new(ExtensibleDataSource::new(ds.clone(), state.clone())); - let mut app = App::<_, Error>::with_state(AppState::from(ExtensibleDataSource::new( - ds, - state.clone(), - ))); - - // Initialize v0 and v1 status API. - register_api("status", &mut app, move |ver| { - status::define_api(&Default::default(), SequencerApiVersion::instance(), ver) - .context("failed to define status api") - })?; - - self.init_hotshot_modules(&mut app)?; - - // Initialize hotshot events API if enabled - if self.hotshot_events.is_some() { - self.init_hotshot_events_module(&mut app)?; - } - drop(app); + let axum_ds = Arc::new(ExtensibleDataSource::new(ds, state.clone())); let port = self.http.port; - let env_vars = endpoints::get_public_env_vars().unwrap_or_default(); + let env_vars = get_public_env_vars().unwrap_or_default(); let node_cfg = self.public_node_config.as_deref().cloned(); let modules = espresso_api::OptionalModules { submit: self.submit.is_some(), @@ -280,18 +244,8 @@ impl Options { // // If we have no availability API, we cannot load a saved leaf from local storage, // so we better have been provided the leaf ahead of time if we want it at all. - let mut app = App::<_, Error>::with_state(AppState::from(state.clone())); - - self.init_hotshot_modules(&mut app)?; - - // Initialize hotshot events API if enabled - if self.hotshot_events.is_some() { - self.init_hotshot_events_module(&mut app)?; - } - drop(app); - let port = self.http.port; - let env_vars = endpoints::get_public_env_vars().unwrap_or_default(); + let env_vars = get_public_env_vars().unwrap_or_default(); let node_cfg = self.public_node_config.as_deref().cloned(); let modules = espresso_api::OptionalModules { submit: self.submit.is_some(), @@ -325,89 +279,12 @@ impl Options { Ok(ctx.with_task_list(tasks)) } - async fn init_app_modules( - &self, - ds: D, - state: ApiState, - bind_version: SequencerApiVersion, - ) -> anyhow::Result<( - Box, - Arc>, - App>, Error>, - )> - where - N: ConnectedNetwork, - P: SequencerPersistence, - D: SequencerDataSource + CatchupStorage + PruningDataSource + Send + Sync + 'static, - { - let metrics = ds.populate_metrics(); - // Deposit the underlying prometheus::Registry for the in-process - // telemetry push task. Idempotent; safe to call multiple times. - telemetry::set_registry(Arc::new(ds.metrics().registry().clone())); - let ds = Arc::new(ExtensibleDataSource::new(ds, state.clone())); - let api_state: endpoints::AvailState = ds.clone().into(); - let mut app = App::<_, Error>::with_state(api_state); - - // Initialize v0 and v1 status API. - register_api("status", &mut app, move |ver| { - status::define_api(&Default::default(), SequencerApiVersion::instance(), ver) - .context("failed to define status api") - })?; - - // Initialize availability and node APIs (these both use the same data source). - - // Note: We initialize two versions of the availability module: `availability/v0` and `availability/v1`. - // - `availability/v0/leaf/0` returns the old `Leaf1` type for backward compatibility. - // - `availability/v1/leaf/0` returns the new `Leaf2` type - - register_api("availability", &mut app, move |ver| { - endpoints::availability(ver).context("failed to define availability api") - })?; - - register_api("node", &mut app, move |ver| { - endpoints::node(ver).context("failed to define node api") - })?; - - register_api("token", &mut app, move |ver| { - endpoints::token(ver).context("failed to define token api") - })?; - - // Initialize submit API - if self.submit.is_some() { - register_api("submit", &mut app, move |ver| { - endpoints::submit::<_, _, _, SequencerApiVersion>(ver) - .context("failed to define submit api") - })?; - } - - tracing::info!("initializing catchup API"); - - register_api("catchup", &mut app, move |ver| { - endpoints::catchup(bind_version, ver).context("failed to define catchup api") - })?; - - register_api("state-signature", &mut app, move |ver| { - endpoints::state_signature(bind_version, ver) - .context("failed to define state signature api") - })?; - - if self.config.is_some() { - let node_cfg = self.public_node_config.as_deref().cloned(); - register_api("config", &mut app, move |ver| { - endpoints::config(bind_version, ver, node_cfg.clone()) - .context("failed to define config api") - })?; - } - Ok((metrics, ds, app)) - } - async fn init_with_query_module_fs( &self, query_opt: Query, mod_opt: persistence::fs::Options, state: ApiState, tasks: &mut TaskList, - bind_version: SequencerApiVersion, ) -> anyhow::Result<( Box, Box, @@ -435,19 +312,11 @@ impl Options { tasks.spawn("process_metrics", ProcessMetrics::new(ds.metrics()).run()); - let (metrics, ds, mut app) = self - .init_app_modules(ds, state.clone(), bind_version) - .await?; - - // Initialize hotshot events API if enabled - if self.hotshot_events.is_some() { - self.init_hotshot_events_module(&mut app)?; - } - drop(app); + let (metrics, ds) = init_query_data_source(ds, state.clone()); let port = self.http.port; let ds_for_axum = ds.clone(); - let env_vars = endpoints::get_public_env_vars().unwrap_or_default(); + let env_vars = get_public_env_vars().unwrap_or_default(); let node_cfg = self.public_node_config.as_deref().cloned(); let modules = espresso_api::OptionalModules { submit: self.submit.is_some(), @@ -484,7 +353,6 @@ impl Options { mod_opt: persistence::sql::Options, state: ApiState, tasks: &mut TaskList, - bind_version: SequencerApiVersion, ) -> anyhow::Result<( Box, Box, @@ -516,55 +384,7 @@ impl Options { let ds = sql::DataSource::create(mod_opt.clone(), provider, false).await?; let inner_storage = ds.inner(); tasks.spawn("process_metrics", ProcessMetrics::new(ds.metrics()).run()); - let (metrics, ds, mut app) = self - .init_app_modules(ds, state.clone(), bind_version) - .await?; - - if self.explorer.is_some() { - register_api("explorer", &mut app, move |ver| { - endpoints::explorer(ver).context("failed to define explorer api") - })?; - } - - // Initialize database metadata API (SQL-only) - register_api("database", &mut app, move |ver| { - endpoints::database::<_, SequencerApiVersion>(ver) - .context("failed to define database api") - })?; - - // Initialize merklized state module for block merkle tree - - register_api("block-state", &mut app, move |ver| { - endpoints::merklized_state::(ver) - .context("failed to define block-state api") - })?; - - // Initialize merklized state module for fee merkle tree - - register_api("fee-state", &mut app, move |ver| { - endpoints::fee::<_, SequencerApiVersion>(ver).context("failed to define fee-state api") - })?; - - register_api("reward-state", &mut app, move |ver| { - endpoints::reward::< - _, - SequencerApiVersion, - RewardMerkleTreeV1, - { RewardMerkleTreeV1::ARITY }, - >(ver, RewardMerkleTreeVersion::V1) - .context("failed to define reward-state api") - })?; - - // register new api for new reward merkle tree - register_api("reward-state-v2", &mut app, move |ver| { - endpoints::reward::< - _, - SequencerApiVersion, - RewardMerkleTreeV2, - { RewardMerkleTreeV2::ARITY }, - >(ver, RewardMerkleTreeVersion::V2) - .context("failed to define reward-state api") - })?; + let (metrics, ds) = init_query_data_source(ds, state.clone()); let get_node_state = { let state = state.clone(); @@ -575,28 +395,9 @@ impl Options { update_state_storage_loop(ds.clone(), get_node_state), ); - // Initialize hotshot events API if enabled - if self.hotshot_events.is_some() { - self.init_hotshot_events_module(&mut app)?; - } - - // Initialize light client API if enabled. - if self.light_client.is_some() { - register_api("light-client", &mut app, move |ver| { - light_client::define_api::<_, SequencerApiVersion>(Default::default(), ver) - .context("failed to define light client api") - })?; - } - - // Drop the tide-disco app — SQL mode is fully served by Axum. The unused `app` here - // is kept above only so the registrations exercise the tide-disco module definitions - // (which still compile against `App::register_module`) until all callers are off - // tide-disco. TODO: stop building `app` once the unused tide-disco branches are gone. - drop(app); - let port = self.http.port; let ds_for_axum = ds.clone(); - let env_vars = endpoints::get_public_env_vars().unwrap_or_default(); + let env_vars = get_public_env_vars().unwrap_or_default(); let node_cfg = self.public_node_config.as_deref().cloned(); let modules = espresso_api::OptionalModules { submit: self.submit.is_some(), @@ -633,105 +434,6 @@ impl Options { Some(RequestResponseStorage::Sql(inner_storage)), )) } - - /// Initialize the modules for interacting with HotShot. - /// - /// This function adds the `submit`, `state`, and `state_signature` API modules to the given - /// app. These modules only require a HotShot handle as state, and thus they work with any data - /// source, so initialization is the same no matter what mode the service is running in. - fn init_hotshot_modules(&self, app: &mut App) -> anyhow::Result<()> - where - S: 'static + Send + Sync + ReadState, - P: SequencerPersistence, - S::State: Send - + Sync - + SubmitDataSource - + StateSignatureDataSource - + NodeStateDataSource - + CatchupDataSource - + HotShotConfigDataSource, - N: ConnectedNetwork, - { - let bind_version = SequencerApiVersion::instance(); - // Initialize submit API - if self.submit.is_some() { - register_api("submit", app, move |ver| { - endpoints::submit::<_, _, _, SequencerApiVersion>(ver) - .context("failed to define submit api") - })?; - } - - // Initialize state API. - if self.catchup.is_some() { - tracing::info!("initializing state API"); - - register_api("catchup", app, move |ver| { - endpoints::catchup(bind_version, ver).context("failed to define catchup api") - })?; - } - - register_api("state-signature", app, move |ver| { - endpoints::state_signature(bind_version, ver) - .context("failed to define state signature api") - })?; - - if self.config.is_some() { - let node_cfg = self.public_node_config.as_deref().cloned(); - register_api("config", app, move |ver| { - endpoints::config(bind_version, ver, node_cfg.clone()) - .context("failed to define config api") - })?; - } - - Ok(()) - } - - /// Initialize the hotshot events API module if enabled. - /// - /// This function adds the hotshot events API module to the given app if the hotshot_events - /// option is enabled. This module requires the app state to implement EventsSource. - fn init_hotshot_events_module(&self, app: &mut App) -> anyhow::Result<()> - where - S: 'static + Send + Sync + ReadState, - S::State: Send + Sync + hotshot_events_service::events_source::EventsSource, - { - tracing::info!("Initializing HotShot events API at /hotshot-events"); - register_api("hotshot-events", app, move |ver| { - hotshot_events_service::events::define_api::<_, _, SequencerApiVersion>( - &hotshot_events_service::events::Options::default(), - ver, - ) - .with_context(|| "failed to define the HotShot events API") - })?; - - Ok(()) - } - - // Kept until tide-disco removal; no longer called since the axum cutover. - #[allow(dead_code)] - fn listen( - &self, - port: u16, - app: App, - bind_version: ApiVer, - ) -> impl Future> + use - where - S: Send + Sync + 'static, - E: Send + Sync + tide_disco::Error, - ApiVer: StaticVersionType + 'static, - { - let max_connections = self.http.max_connections; - - async move { - if let Some(limit) = max_connections { - app.serve(RateLimitListener::with_port(port, limit), bind_version) - .await?; - } else { - app.serve(format!("0.0.0.0:{port}"), bind_version).await?; - } - Ok(()) - } - } } /// The minimal HTTP API. @@ -823,26 +525,43 @@ pub struct Explorer; #[derive(Parser, Clone, Copy, Debug, Default)] pub struct LightClient; -/// Registers two versions (v0 and v1) of the same API module under the given path. -fn register_api( - path: &'static str, - app: &mut App, - f: F, -) -> anyhow::Result<()> +/// Metrics handle plus the wrapped query data source shared by the axum server and update loops. +type QueryModuleState = (Box, Arc>); + +/// Populate consensus metrics on `ds`, deposit its prometheus registry for the in-process +/// telemetry push task (idempotent), and wrap it with the API state. +fn init_query_data_source(ds: D, state: ApiState) -> QueryModuleState where - S: 'static + Send + Sync, - E: Send + Sync + 'static + tide_disco::Error + From, - ModuleError: Send + Sync + 'static, - ModuleVersion: StaticVersionType + 'static, - F: Fn(semver::Version) -> anyhow::Result>, + N: ConnectedNetwork, + P: SequencerPersistence, + D: SequencerDataSource + CatchupStorage + PruningDataSource + Send + Sync + 'static, { - let v0 = "0.0.1".parse().unwrap(); - let v1 = "1.1.0".parse().unwrap(); - let result1 = f(v0)?; - let result2 = f(v1)?; + let metrics = ds.populate_metrics(); + telemetry::set_registry(Arc::new(ds.metrics().registry().clone())); + let ds = Arc::new(ExtensibleDataSource::new(ds, state)); + (metrics, ds) +} - app.register_module(path, result1)?; - app.register_module(path, result2)?; +/// The environment variables listed in `api/public-env-vars.toml`, as `KEY=value` strings. +fn get_public_env_vars() -> anyhow::Result> { + let toml: toml::Value = toml::from_str(include_str!("../../api/public-env-vars.toml"))?; + + let keys = toml + .get("variables") + .ok_or_else(|| toml::de::Error::custom("variables not found"))? + .as_array() + .ok_or_else(|| toml::de::Error::custom("variables is not an array"))? + .clone() + .into_iter() + .map(|v| v.try_into()) + .collect::, toml::de::Error>>()?; + + let hashmap: HashMap = env::vars().collect(); + let mut public_env_vars: Vec = Vec::new(); + for key in keys { + let value = hashmap.get(&key).cloned().unwrap_or_default(); + public_env_vars.push(format!("{key}={value}")); + } - Ok(()) + Ok(public_env_vars) } diff --git a/crates/espresso/node/src/api/state.rs b/crates/espresso/node/src/api/state.rs index 6880e625aa2..6b88b85780b 100644 --- a/crates/espresso/node/src/api/state.rs +++ b/crates/espresso/node/src/api/state.rs @@ -8,6 +8,7 @@ use std::{ops::Bound, time::Duration}; use alloy::primitives::U256; use async_trait::async_trait; use committable::Committable as _; +use disco_types::{error::Error as _, status::StatusCode}; use espresso_api::{error::AvailabilityError, v1::HotShotAvailabilityApi}; use espresso_types::{ NamespaceId, NamespaceProofQueryData, NsProof, SeqTypes, @@ -41,7 +42,7 @@ use hotshot_query_service::{ MerklizedStateDataSource, MerklizedStateHeightPersistence, Snapshot as HsSnapshot, }, node::{NodeDataSource as _, WindowStart}, - status::HasMetrics, + status::HasMetrics as _, types::HeightIndexed as _, }; use hotshot_types::{ @@ -53,6 +54,7 @@ use jf_merkle_tree_compat::prelude::{ MerkleNode as InternalMerkleNode, MerkleProof as InternalMerkleProof, MerkleProof as JfMerkleProof, }; +use prometheus::Encoder as _; use serde_json; use serialization_api::v2::{ self, RewardAccountProofV2, RewardAccountQueryDataV2, RewardBalance, RewardBalances, @@ -60,7 +62,6 @@ use serialization_api::v2::{ reward_merkle_proof_v2::ProofType, }; use tagged_base64::TaggedBase64; -use tide_disco::{Error as _, StatusCode, metrics::Metrics as _}; use super::{ RewardMerkleTreeDataSource, RewardMerkleTreeV2Data as InternalRewardTreeData, @@ -2308,7 +2309,10 @@ where async fn metrics(&self) -> anyhow::Result { let ds = &*self.data_source; - ds.metrics().export().map_err(|e| anyhow::anyhow!("{e}")) + // Same text exposition tide-disco's `Metrics::export` produced for this registry. + let mut buffer = Vec::new(); + prometheus::TextEncoder::new().encode(&ds.metrics().registry().gather(), &mut buffer)?; + Ok(String::from_utf8(buffer)?) } } diff --git a/crates/espresso/node/src/state_cert.rs b/crates/espresso/node/src/state_cert.rs index f4155ba9856..8ae9728bc02 100644 --- a/crates/espresso/node/src/state_cert.rs +++ b/crates/espresso/node/src/state_cert.rs @@ -4,6 +4,7 @@ use std::collections::HashMap; use alloy::primitives::{FixedBytes, U256}; use anyhow::bail; +use disco_types::status::StatusCode; use espresso_types::SeqTypes; use hotshot_contract_adapter::light_client::derive_signed_state_digest; use hotshot_query_service::availability::Error; @@ -13,7 +14,6 @@ use hotshot_types::{ stake_table::HSStakeTable, traits::signature_key::{LCV2StateSignatureKey, LCV3StateSignatureKey, StakeTableEntryType}, }; -use tide_disco::StatusCode; /// Error type for state certificate fetching #[derive(Debug, thiserror::Error)] diff --git a/crates/hotshot/orchestrator/Cargo.toml b/crates/hotshot/orchestrator/Cargo.toml index dc50c4ed9f4..fdf1b64816d 100644 --- a/crates/hotshot/orchestrator/Cargo.toml +++ b/crates/hotshot/orchestrator/Cargo.toml @@ -12,6 +12,9 @@ axum = { workspace = true } blake3 = { workspace = true } clap = { workspace = true } csv = { workspace = true } +# Only used for its `error::ServerError` wire-error type: the `OrchestratorApi` trait's handler +# methods are still expressed in terms of it. The HTTP server itself is axum. +disco-types = { workspace = true } futures = { workspace = true } hotshot-types = { workspace = true } http-client = { workspace = true } @@ -19,9 +22,6 @@ libp2p-identity = { workspace = true } multiaddr = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } -# Only used for its `error::ServerError` wire-error type: the `OrchestratorApi` trait's handler -# methods are still expressed in terms of it. The HTTP server itself is axum. -tide-disco = { workspace = true } tokio = { workspace = true } toml = { workspace = true } tracing = { workspace = true } diff --git a/crates/hotshot/orchestrator/src/lib.rs b/crates/hotshot/orchestrator/src/lib.rs index e0bd968e1d7..b8d4f3fd2b2 100644 --- a/crates/hotshot/orchestrator/src/lib.rs +++ b/crates/hotshot/orchestrator/src/lib.rs @@ -30,6 +30,7 @@ use axum::{ }; use client::{BenchResults, BenchResultsDownloadConfig}; use csv::Writer; +use disco_types::error::ServerError; use futures::{StreamExt, stream::FuturesUnordered}; use hotshot_types::{ PeerConfig, @@ -46,7 +47,6 @@ use libp2p_identity::{ }; use multiaddr::Multiaddr; use serde::{Serialize, de::DeserializeOwned}; -use tide_disco::error::ServerError; use tokio::net::TcpListener; use vbs::{BinarySerializer, Serializer, version::StaticVersion}; @@ -264,7 +264,7 @@ where if !self.accepting_new_keys { return Err(ServerError { - status: tide_disco::StatusCode::FORBIDDEN, + status: disco_types::status::StatusCode::FORBIDDEN, message: "Network has been started manually, and is no longer registering new \ keys." .to_string(), @@ -351,7 +351,7 @@ where }) else { return Err(ServerError { - status: tide_disco::StatusCode::FORBIDDEN, + status: disco_types::status::StatusCode::FORBIDDEN, message: "You are unauthorized to register with the orchestrator".to_string(), }); }; @@ -359,7 +359,7 @@ where // Check that our recorded DA status for the node matches what the node actually requested if node_config.da != da_requested { return Err(ServerError { - status: tide_disco::StatusCode::BAD_REQUEST, + status: disco_types::status::StatusCode::BAD_REQUEST, message: format!( "Mismatch in DA status in registration for node {}. DA requested: {}, \ expected: {}", @@ -412,7 +412,7 @@ where if usize::from(node_index) >= self.config.config.num_nodes_with_stake.get() { return Err(ServerError { - status: tide_disco::StatusCode::BAD_REQUEST, + status: disco_types::status::StatusCode::BAD_REQUEST, message: "Network has reached capacity".to_string(), }); } @@ -447,7 +447,7 @@ where if usize::from(tmp_node_index) >= self.config.config.num_nodes_with_stake.get() { return Err(ServerError { - status: tide_disco::StatusCode::BAD_REQUEST, + status: disco_types::status::StatusCode::BAD_REQUEST, message: "Node index getter for key pair generation has reached capacity" .to_string(), }); @@ -472,7 +472,7 @@ where fn peer_pub_ready(&self) -> Result { if !self.peer_pub_ready { return Err(ServerError { - status: tide_disco::StatusCode::BAD_REQUEST, + status: disco_types::status::StatusCode::BAD_REQUEST, message: "Peer's public configs are not ready".to_string(), }); } @@ -482,7 +482,7 @@ where fn post_config_after_peer_collected(&mut self) -> Result, ServerError> { if !self.peer_pub_ready { return Err(ServerError { - status: tide_disco::StatusCode::BAD_REQUEST, + status: disco_types::status::StatusCode::BAD_REQUEST, message: "Peer's public configs are not ready".to_string(), }); } @@ -494,7 +494,7 @@ where // println!("{}", self.start); if !self.start { return Err(ServerError { - status: tide_disco::StatusCode::BAD_REQUEST, + status: disco_types::status::StatusCode::BAD_REQUEST, message: "Network is not ready to start".to_string(), }); } @@ -512,7 +512,7 @@ where .contains(peer_config) { return Err(ServerError { - status: tide_disco::StatusCode::FORBIDDEN, + status: disco_types::status::StatusCode::FORBIDDEN, message: "You are unauthorized to register with the orchestrator".to_string(), }); } @@ -542,7 +542,7 @@ where fn post_manual_start(&mut self, password_bytes: Vec) -> Result<(), ServerError> { if !self.manual_start_allowed { return Err(ServerError { - status: tide_disco::StatusCode::FORBIDDEN, + status: disco_types::status::StatusCode::FORBIDDEN, message: "Configs have already been distributed to nodes, and the network can no \ longer be started manually." .to_string(), @@ -555,7 +555,7 @@ where // Check that the password matches if self.config.manual_start_password != Some(password) { return Err(ServerError { - status: tide_disco::StatusCode::FORBIDDEN, + status: disco_types::status::StatusCode::FORBIDDEN, message: "Incorrect password.".to_string(), }); } @@ -571,7 +571,7 @@ where self.config.config.da_staked_committee_size = registered_da_nodes; } else { return Err(ServerError { - status: tide_disco::StatusCode::FORBIDDEN, + status: disco_types::status::StatusCode::FORBIDDEN, message: format!( "We cannot manually start the network, because we only have \ {registered_nodes_with_stake} nodes with stake registered, with \ @@ -663,7 +663,7 @@ where && self.builders.len() != self.config.config.da_staked_committee_size { return Err(ServerError { - status: tide_disco::StatusCode::NOT_FOUND, + status: disco_types::status::StatusCode::NOT_FOUND, message: "Not all builders are registered yet".to_string(), }); } @@ -681,7 +681,7 @@ fn wants_binary(headers: &HeaderMap) -> bool { .is_some_and(|v| v.contains("application/octet-stream")) } -fn axum_status(status: tide_disco::StatusCode) -> StatusCode { +fn axum_status(status: disco_types::status::StatusCode) -> StatusCode { StatusCode::from_u16(u16::from(status)).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) } @@ -697,7 +697,7 @@ fn encode_ok(headers: &HeaderMap, value: T) -> Response { Err(err) => encode_err( headers, ServerError { - status: tide_disco::StatusCode::INTERNAL_SERVER_ERROR, + status: disco_types::status::StatusCode::INTERNAL_SERVER_ERROR, message: err.to_string(), }, ), @@ -734,7 +734,7 @@ fn respond(headers: &HeaderMap, result: Result) -> fn malformed_body() -> ServerError { ServerError { - status: tide_disco::StatusCode::BAD_REQUEST, + status: disco_types::status::StatusCode::BAD_REQUEST, message: "Malformed body".to_string(), } } @@ -895,7 +895,7 @@ async fn post_builder( match reachable.next().await { Some(url) => state.write().await.post_builder(url), None => Err(ServerError { - status: tide_disco::StatusCode::BAD_REQUEST, + status: disco_types::status::StatusCode::BAD_REQUEST, message: "No reachable addresses".to_string(), }), } diff --git a/light-client-query-service/Cargo.toml b/light-client-query-service/Cargo.toml index cf8c0361501..31ef730db8e 100644 --- a/light-client-query-service/Cargo.toml +++ b/light-client-query-service/Cargo.toml @@ -18,6 +18,7 @@ genesis = [] anyhow = { workspace = true } axum = { workspace = true } clap = { workspace = true } +disco-types = { workspace = true } espresso-node = { workspace = true, default-features = false } espresso-types = { workspace = true, features = ["node"] } futures = { workspace = true } @@ -28,7 +29,6 @@ light-client = { workspace = true, features = ["client", "rlp"] } log-panics = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } -tide-disco = { workspace = true } tokio = { workspace = true } toml = { workspace = true } tracing = { workspace = true } diff --git a/light-client-query-service/src/api.rs b/light-client-query-service/src/api.rs index 54ff118a2f7..6eb7b0bd0e4 100644 --- a/light-client-query-service/src/api.rs +++ b/light-client-query-service/src/api.rs @@ -1,5 +1,5 @@ //! Axum port of the `availability` and `node` API modules that this service used to serve via -//! `hotshot_query_service::{availability, node}::define_api` on a `tide_disco::App`. +//! `hotshot_query_service::{availability, node}::define_api` on a tide-disco `App`. //! //! Route paths, status codes and the wire error type are all taken directly from //! `hotshot-query-service` (see `availability.rs`/`node.rs` there and their handler bodies) so @@ -17,6 +17,7 @@ use axum::{ response::{IntoResponse, Response}, routing::get, }; +use disco_types::error::Error as _; use espresso_node::api::sql::DataSource; use espresso_types::SeqTypes; use futures::{StreamExt as _, TryStreamExt as _, stream::BoxStream}; @@ -35,7 +36,6 @@ use hotshot_query_service::{ use hotshot_types::data::VidCommitment; use http_client::healthcheck::HealthStatus; use serde::Serialize; -use tide_disco::Error as _; use vbs::{BinarySerializer, Serializer, version::StaticVersion}; /// Binary framing version for VBS-negotiated responses, matching the wire version this service @@ -57,7 +57,7 @@ fn wants_binary(headers: &HeaderMap) -> bool { } /// Maps `hotshot_query_service`'s wrapped `reqwest`-based status code onto axum's. -fn wire_status(status: tide_disco::StatusCode) -> StatusCode { +fn wire_status(status: disco_types::status::StatusCode) -> StatusCode { StatusCode::from_u16(u16::from(status)).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) } @@ -115,7 +115,7 @@ where { value.parse().map_err(|e| availability::Error::Custom { message: format!("invalid {field}: {e}"), - status: tide_disco::StatusCode::BAD_REQUEST, + status: disco_types::status::StatusCode::BAD_REQUEST, }) } @@ -126,7 +126,7 @@ where { value.parse().map_err(|e| node::Error::Custom { message: format!("invalid {field}: {e}"), - status: tide_disco::StatusCode::BAD_REQUEST, + status: disco_types::status::StatusCode::BAD_REQUEST, }) }