espresso_api/

lib.rs

//! Espresso API server with both Axum (HTTP/JSON) and gRPC endpoints

// Module declarations
mod axum;
pub mod error;
mod render;
pub mod v1;

/// The v2 API contract, generated by `build.rs` from the proto files in `proto/`: message types,
/// the tonic server traits, and the protoJSON `Serialize`/`Deserialize` impls that define the
/// HTTP wire encoding. `render` builds these messages from consensus types.
pub mod proto {
    // Every pbjson `Deserialize` impl formats its field list as `{:?}` through a reference.
    #![allow(clippy::useless_borrows_in_formatting)]
    #![expect(
        clippy::large_enum_variant,
        reason = "prost lays every oneof arm out inline, so the ADVZ transaction proof dwarfs its \
                  siblings, and boxing it (`Builder::boxed`) breaks pbjson-build's serde"
    )]

    include!(concat!(env!("OUT_DIR"), "/espresso.api.v2.rs"));
    include!(concat!(env!("OUT_DIR"), "/espresso.api.v2.serde.rs"));
}

/// Axum REST handlers derived from the `google.api.http` annotations, transcoding
/// HTTP/JSON onto the tonic service traits.
pub mod rest {
    // The generator emits `#[expect]` attributes that not every handler fulfills.
    #![allow(unfulfilled_lint_expectations)]
    #![expect(
        clippy::too_many_arguments,
        reason = "`all_rest_routes` takes one service per proto file, so its arity grows with the \
                  API"
    )]

    include!(concat!(env!("OUT_DIR"), "/espresso.api.v2.rest.rs"));
}

/// The compiled proto descriptor set, for gRPC reflection.
pub const FILE_DESCRIPTOR_SET: &[u8] = include_bytes!(concat!(env!("OUT_DIR"), "/descriptor.bin"));

use tower::Layer;

// Re-exports
pub use self::axum::{create_router_v1, routes};
use self::proto::{
    availability_service_server::{AvailabilityService, AvailabilityServiceServer},
    catchup_service_server::{CatchupService, CatchupServiceServer},
    config_service_server::{ConfigService, ConfigServiceServer},
    database_service_server::{DatabaseService, DatabaseServiceServer},
    explorer_service_server::{ExplorerService, ExplorerServiceServer},
    light_client_service_server::{LightClientService, LightClientServiceServer},
    merklized_state_service_server::{MerklizedStateService, MerklizedStateServiceServer},
    node_service_server::{NodeService, NodeServiceServer},
    reward_state_service_server::{RewardStateService, RewardStateServiceServer},
    state_signature_service_server::{StateSignatureService, StateSignatureServiceServer},
    status_service_server::{StatusService, StatusServiceServer},
    submit_service_server::{SubmitService, SubmitServiceServer},
    token_service_server::{TokenService, TokenServiceServer},
};

/// Build a full request URL from a server base URL and a path produced by one of the
/// `routes::v1::*` builders.
///
/// Use this from test/CLI sites that have a `url::Url` pointing at the API server and want
/// the absolute URL for a single request. Internally this is just `base.join(path)`; the
/// helper exists so the (path-const, builder, joiner) trio reads as one chain.
pub fn url(base: &::url::Url, path: impl AsRef<str>) -> ::url::Url {
    base.join(path.as_ref())
        .expect("path produced by routes::*::path_fn is always a valid relative URL")
}

/// Start Axum HTTP server with combined v1 and v2 APIs
///
/// This serves both APIs at /v1/* and /v2/* from a single state implementation.
///
/// `catchup`, like the query-service modules (`status`, `availability`, `node`, `token`,
/// `block-state`, `fee-state`, `reward-state`, `database`) and `v2`, is always on: tide-disco's
/// SQL mode registered it unconditionally. `light-client` is always on too, since peers fetch
/// missing data from it. `submit`, `config`, and `explorer` follow `Options`, matching
/// `Options::init_with_query_module_sql`, and v2's gated services follow the same flag as their
/// v1 module.
pub async fn serve_axum<S>(
    port: u16,
    state: S,
    modules: OptionalModules,
    max_connections: Option<usize>,
) -> anyhow::Result<()>
where
    S: v1::RewardApi
        + v1::AvailabilityApi
        + v1::HotShotAvailabilityApi
        + v1::BlockStateApi
        + v1::FeeStateApi
        + v1::StatusApi
        + v1::ConfigApi
        + v1::NodeApi
        + v1::CatchupApi
        + v1::SubmitApi
        + v1::StateSignatureApi
        + v1::LightClientApi
        + v1::ExplorerApi
        + v1::TokenApi
        + v1::DatabaseApi
        + StatusService
        + TokenService
        + NodeService
        + ConfigService
        + DatabaseService
        + AvailabilityService
        + MerklizedStateService
        + RewardStateService
        + StateSignatureService
        + SubmitService
        + ExplorerService
        + CatchupService
        + LightClientService
        + Send
        + Sync
        + 'static,
{
    let listener = bind_api(port).await?;
    let state = std::sync::Arc::new(state);
    let mut router = axum::router_reward(state.clone())
        .merge(axum::router_availability(state.clone()))
        .merge(axum::router_block_state(state.clone()))
        .merge(axum::router_fee_state(state.clone()))
        .merge(axum::router_status(state.clone()))
        .merge(axum::router_node(state.clone()))
        .merge(axum::router_catchup(state.clone()))
        .merge(axum::router_state_signature(state.clone()))
        .merge(axum::router_token(state.clone()))
        .merge(axum::router_database(state.clone()))
        .merge(axum::router_light_client(state.clone()));
    if modules.submit {
        router = router.merge(axum::router_submit(state.clone()));
    }
    if modules.config {
        router = router.merge(axum::router_config(state.clone()));
    }
    if modules.explorer {
        router = router.merge(axum::router_explorer(state.clone()));
    }

    let router = axum::finish_v1_docs(router)
        .merge(router_v2(state, modules))
        .merge(axum::router_v2_docs());
    serve_router(listener, "v1 and v2", router, max_connections).await
}

/// The v2 REST routes exactly as [`serve_axum`] mounts them. Extracted so the test asserting
/// every documented route is mounted exercises the same construction; don't inline it back.
pub(crate) fn router_v2<S>(state: std::sync::Arc<S>, modules: OptionalModules) -> ::axum::Router
where
    S: StatusService
        + TokenService
        + NodeService
        + ConfigService
        + DatabaseService
        + AvailabilityService
        + MerklizedStateService
        + RewardStateService
        + StateSignatureService
        + SubmitService
        + ExplorerService
        + CatchupService
        + LightClientService
        + Send
        + Sync
        + 'static,
{
    let router = rest::status_service_rest_router(state.clone())
        .merge(rest::token_service_rest_router(state.clone()))
        .merge(rest::node_service_rest_router(state.clone()))
        .merge(rest::database_service_rest_router(state.clone()))
        .merge(rest::availability_service_rest_router(state.clone()))
        .merge(rest::merklized_state_service_rest_router(state.clone()))
        .merge(rest::reward_state_service_rest_router(state.clone()))
        .merge(rest::state_signature_service_rest_router(state.clone()))
        .merge(rest::catchup_service_rest_router(state.clone()))
        .merge(rest::light_client_service_rest_router(state.clone()));
    let router = if modules.submit {
        router.merge(rest::submit_service_rest_router(state.clone()))
    } else {
        router.merge(axum::router_module_disabled(
            "submit",
            routes::v2::SUBMIT_ROUTES,
        ))
    };
    let router = if modules.explorer {
        router.merge(rest::explorer_service_rest_router(state.clone()))
    } else {
        router.merge(axum::router_module_disabled(
            "explorer",
            routes::v2::EXPLORER_ROUTES,
        ))
    };
    let router = if modules.config {
        router.merge(rest::config_service_rest_router(state))
    } else {
        router.merge(axum::router_module_disabled(
            "config",
            routes::v2::CONFIG_ROUTES,
        ))
    };
    router
        .layer(::axum::middleware::from_fn(axum::v2_refuse_post_query))
        .layer(::axum::middleware::from_fn(axum::v2_error_envelope))
}

/// Which of the optional API modules to serve, for modes that make them conditional
/// (mirroring `Options::submit`/`Options::config`/`Options::explorer`).
#[derive(Default, Clone, Copy, Debug)]
pub struct OptionalModules {
    pub submit: bool,
    pub catchup: bool,
    pub config: bool,
    pub explorer: bool,
}

/// Serve the query API used by the filesystem-backed storage mode: status, availability, node,
/// token, catchup, and state-signature are always on (tide registered them unconditionally);
/// submit and config follow `Options`. Filesystem storage doesn't implement the
/// reward/merklized-state/explorer/database traits, so those modules aren't served (a request to
/// one of their routes 404s, matching tide).
pub async fn serve_axum_fs<S>(
    port: u16,
    state: S,
    modules: OptionalModules,
    max_connections: Option<usize>,
) -> anyhow::Result<()>
where
    S: v1::StatusApi
        + v1::AvailabilityApi
        + v1::HotShotAvailabilityApi
        + v1::NodeApi
        + v1::TokenApi
        + v1::CatchupApi
        + v1::SubmitApi
        + v1::StateSignatureApi
        + v1::ConfigApi
        + Send
        + Sync
        + 'static,
{
    let listener = bind_api(port).await?;
    let state = std::sync::Arc::new(state);
    let mut router = axum::router_status(state.clone())
        .merge(axum::router_availability(state.clone()))
        .merge(axum::router_node(state.clone()))
        .merge(axum::router_token(state.clone()))
        .merge(axum::router_catchup(state.clone()))
        .merge(axum::router_state_signature(state.clone()));
    if modules.submit {
        router = router.merge(axum::router_submit(state.clone()));
    }
    if modules.config {
        router = router.merge(axum::router_config(state.clone()));
    }

    serve_router(
        listener,
        "fs",
        axum::finish_v1_docs(router),
        max_connections,
    )
    .await
}

/// Serve the status-only API: no availability/node/token data source is available, so only
/// status and the HotShot modules (submit, catchup, state-signature, config) can
/// be served. State-signature is always on; the rest follow `Options`.
pub async fn serve_axum_status<S>(
    port: u16,
    state: S,
    modules: OptionalModules,
    max_connections: Option<usize>,
) -> anyhow::Result<()>
where
    S: v1::StatusApi
        + v1::SubmitApi
        + v1::CatchupApi
        + v1::StateSignatureApi
        + v1::ConfigApi
        + Send
        + Sync
        + 'static,
{
    let listener = bind_api(port).await?;
    let state = std::sync::Arc::new(state);
    let router =
        axum::router_status(state.clone()).merge(axum::router_state_signature(state.clone()));
    let router = merge_hotshot_modules(router, &state, modules);
    serve_router(
        listener,
        "status",
        axum::finish_v1_docs(router),
        max_connections,
    )
    .await
}

/// Serve the bare API (no query or status module): only the HotShot modules are available,
/// since the only app state is the HotShot handle. State-signature is always on; the rest follow
/// `Options`, matching `Options::init_hotshot_modules`.
pub async fn serve_axum_bare<S>(
    port: u16,
    state: S,
    modules: OptionalModules,
    max_connections: Option<usize>,
) -> anyhow::Result<()>
where
    S: v1::SubmitApi
        + v1::CatchupApi
        + v1::StateSignatureApi
        + v1::ConfigApi
        + Send
        + Sync
        + 'static,
{
    let listener = bind_api(port).await?;
    let state = std::sync::Arc::new(state);
    let router = axum::router_state_signature(state.clone());
    let router = merge_hotshot_modules(router, &state, modules);
    serve_router(
        listener,
        "bare",
        axum::finish_v1_docs(router),
        max_connections,
    )
    .await
}

fn merge_hotshot_modules<S>(
    mut router: aide::axum::ApiRouter,
    state: &std::sync::Arc<S>,
    modules: OptionalModules,
) -> aide::axum::ApiRouter
where
    S: v1::SubmitApi + v1::CatchupApi + v1::ConfigApi + Send + Sync + 'static,
{
    if modules.submit {
        router = router.merge(axum::router_submit(state.clone()));
    }
    if modules.catchup {
        router = router.merge(axum::router_catchup(state.clone()));
    }
    if modules.config {
        router = router.merge(axum::router_config(state.clone()));
    }

    router
}

/// Add the reserved top-level routes, apply the optional concurrency limit, rewrite legacy URIs,
/// and bind/serve the router. Shared by all `serve_axum*` entry points.
/// Bind before composing routers: OpenAPI generation takes ~0.5s in debug builds, and clients
/// connecting during it should queue in the accept backlog rather than get refused.
async fn bind_api(port: u16) -> anyhow::Result<tokio::net::TcpListener> {
    let addr = format!("0.0.0.0:{}", port);
    tracing::info!("Binding to {}", addr);
    Ok(tokio::net::TcpListener::bind(&addr).await?)
}

async fn serve_router(
    listener: tokio::net::TcpListener,
    mode: &str,
    router: ::axum::Router,
    max_connections: Option<usize>,
) -> anyhow::Result<()> {
    let mut router = axum::with_top_level_routes(router);
    if let Some(limit) = max_connections {
        router = apply_connection_limit(router, limit);
    }
    let router = router.layer(http_wire::body_limit_layer());
    // CORS goes on last so it wraps the connection limit, whose 429 would otherwise skip it.
    let router = router.layer(http_wire::cors_layer());
    // `Router::layer` middleware runs after routing, so it can't rewrite a URI to match a
    // different route. Wrapping the whole router with `MapRequestLayer` instead runs the
    // rewrite before routing, per the axum-documented pattern for this case.
    let router = tower::util::MapRequestLayer::new(axum::rewrite_legacy_uri).layer(router);

    tracing::info!(
        "Axum API server listening on {:?} ({} mode)",
        listener.local_addr()?,
        mode
    );
    ::axum::serve(listener, ::axum::ServiceExt::into_make_service(router)).await?;

    tracing::info!("Axum server stopped");
    Ok(())
}

/// Shared budget: a request holds a slot until its handler returns, excess gets 429. A stream
/// (websocket or SSE) releases its slot once the response starts, so open streams are not capped
/// here, see [`axum::limit_requests`].
fn apply_connection_limit(router: ::axum::Router, limit: usize) -> ::axum::Router {
    let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(limit));
    router
        .layer(::axum::middleware::from_fn(axum::limit_requests))
        .layer(::axum::Extension(axum::RequestLimit(semaphore)))
}

/// Start Tonic gRPC server. The gated services follow their module flag like the REST side; the
/// reflection service still lists them, so a disabled deployment answers them with `Unimplemented`.
pub async fn serve_tonic<S>(port: u16, state: S, modules: OptionalModules) -> anyhow::Result<()>
where
    S: StatusService
        + TokenService
        + NodeService
        + ConfigService
        + DatabaseService
        + AvailabilityService
        + MerklizedStateService
        + RewardStateService
        + StateSignatureService
        + SubmitService
        + ExplorerService
        + CatchupService
        + LightClientService
        + Clone,
{
    use ::tonic::transport::Server;

    let addr = std::net::SocketAddr::from(([0, 0, 0, 0], port));

    // Enable gRPC reflection for tools like grpcurl
    let reflection_service = tonic_reflection::server::Builder::configure()
        .register_encoded_file_descriptor_set(FILE_DESCRIPTOR_SET)
        .build_v1()?;

    let router = Server::builder()
        .add_service(StatusServiceServer::new(state.clone()))
        .add_service(TokenServiceServer::new(state.clone()))
        .add_service(NodeServiceServer::new(state.clone()))
        .add_service(DatabaseServiceServer::new(state.clone()))
        .add_service(AvailabilityServiceServer::new(state.clone()))
        .add_service(MerklizedStateServiceServer::new(state.clone()))
        .add_service(RewardStateServiceServer::new(state.clone()))
        .add_service(StateSignatureServiceServer::new(state.clone()))
        .add_service(CatchupServiceServer::new(state.clone()))
        .add_service(LightClientServiceServer::new(state.clone()))
        .add_service(reflection_service)
        .add_optional_service(modules.submit.then(|| {
            // tonic's 4 MiB default would refuse a transaction the REST route and the block size
            // limit accept.
            SubmitServiceServer::new(state.clone())
                .max_decoding_message_size(http_wire::MAX_REQUEST_BODY_BYTES)
        }))
        .add_optional_service(
            modules
                .explorer
                .then(|| ExplorerServiceServer::new(state.clone())),
        )
        .add_optional_service(modules.config.then(|| ConfigServiceServer::new(state)));

    tracing::info!("gRPC server listening on {}", addr);
    router.serve(addr).await?;

    Ok(())
}