espresso_node/api/
options.rs

1//! Sequencer-specific API options and initialization.
2
3use std::{collections::BTreeSet, env, sync::Arc};
4
5use ::light_client::{state::LightClientOptions, storage::LightClientSqliteOptions};
6use anyhow::{Context, bail};
7use clap::Parser;
8use espresso_telemetry as telemetry;
9use espresso_types::{
10    PubKey,
11    v0::traits::{EventConsumer, NullEventConsumer, PersistenceOptions, SequencerPersistence},
12};
13use futures::{channel::oneshot, future::BoxFuture};
14use hotshot_query_service::{
15    data_source::{ExtensibleDataSource, MetricsDataSource},
16    status::{HasMetrics, UpdateStatusData},
17};
18use hotshot_types::traits::{
19    metrics::{Metrics, NoMetrics},
20    network::ConnectedNetwork,
21};
22use process_metrics::ProcessMetrics;
23use url::Url;
24
25use super::{
26    ApiState, StorageState,
27    data_source::{
28        NodeStateDataSource, Provider, PruningDataSource, SequencerDataSource, provider,
29    },
30    fs, sql,
31    state::NodeApiStateImpl,
32    update::ApiEventConsumer,
33};
34use crate::{
35    api::LightClientProvider,
36    catchup::CatchupStorage,
37    context::{SequencerContext, TaskList},
38    options::PublicNodeConfig,
39    persistence,
40    request_response::data_source::Storage as RequestResponseStorage,
41    state::update_state_storage_loop,
42};
43
44#[derive(Clone, Debug)]
45pub struct Options {
46    pub http: Http,
47    pub query: Option<Query>,
48    pub submit: Option<Submit>,
49    pub status: Option<Status>,
50    pub catchup: Option<Catchup>,
51    pub config: Option<Config>,
52    pub hotshot_events: Option<HotshotEvents>,
53    pub explorer: Option<Explorer>,
54    pub light_client: Option<LightClient>,
55    pub storage_fs: Option<persistence::fs::Options>,
56    pub storage_sql: Option<persistence::sql::Options>,
57    pub public_node_config: Option<Box<PublicNodeConfig>>,
58}
59
60impl From<Http> for Options {
61    fn from(http: Http) -> Self {
62        Self {
63            http,
64            query: None,
65            submit: None,
66            status: None,
67            catchup: None,
68            config: None,
69            hotshot_events: None,
70            explorer: None,
71            light_client: None,
72            storage_fs: None,
73            storage_sql: None,
74            public_node_config: None,
75        }
76    }
77}
78
79impl Options {
80    /// Default options for running a web server on the given port.
81    pub fn with_port(port: u16) -> Self {
82        Http::with_port(port).into()
83    }
84
85    /// Add a query API module backed by a Postgres database.
86    pub fn query_sql(mut self, query: Query, storage: persistence::sql::Options) -> Self {
87        self.query = Some(query);
88        self.storage_sql = Some(storage);
89        self
90    }
91
92    /// Add a query API module backed by the file system.
93    pub fn query_fs(mut self, query: Query, storage: persistence::fs::Options) -> Self {
94        self.query = Some(query);
95        self.storage_fs = Some(storage);
96        self
97    }
98
99    /// Add a submit API module.
100    pub fn submit(mut self, opt: Submit) -> Self {
101        self.submit = Some(opt);
102        self
103    }
104
105    /// Add a status API module.
106    pub fn status(mut self, opt: Status) -> Self {
107        self.status = Some(opt);
108        self
109    }
110
111    /// Add a catchup API module.
112    pub fn catchup(mut self, opt: Catchup) -> Self {
113        self.catchup = Some(opt);
114        self
115    }
116
117    /// Add a config API module.
118    pub fn config(mut self, opt: Config) -> Self {
119        self.config = Some(opt);
120        self
121    }
122
123    /// Set the merged runtime configuration exposed via `GET /config/runtime`.
124    ///
125    /// If unset, the `/config/runtime` route returns 404.
126    pub fn public_node_config(mut self, c: PublicNodeConfig) -> Self {
127        self.public_node_config = Some(Box::new(c));
128        self
129    }
130
131    /// Add a Hotshot events streaming API module.
132    pub fn hotshot_events(mut self, opt: HotshotEvents) -> Self {
133        self.hotshot_events = Some(opt);
134        self
135    }
136
137    /// Add an explorer API module.
138    pub fn explorer(mut self, opt: Explorer) -> Self {
139        self.explorer = Some(opt);
140        self
141    }
142
143    /// Add a light client API module.
144    pub fn light_client(mut self, opt: LightClient) -> Self {
145        self.light_client = Some(opt);
146        self
147    }
148
149    /// Whether these options will run the query API.
150    pub fn has_query_module(&self) -> bool {
151        self.query.is_some() && (self.storage_fs.is_some() || self.storage_sql.is_some())
152    }
153
154    /// Start the server.
155    ///
156    /// The function `init_context` is used to create a sequencer context from a metrics object and
157    /// optional saved consensus state. The metrics object is created from the API data source, so
158    /// that consensus will populuate metrics that can then be read and served by the API.
159    pub async fn serve<N, P, F>(mut self, init_context: F) -> anyhow::Result<SequencerContext<N, P>>
160    where
161        N: ConnectedNetwork<PubKey>,
162        P: SequencerPersistence,
163        F: FnOnce(
164            Box<dyn Metrics>,
165            Box<dyn EventConsumer>,
166            Option<RequestResponseStorage>,
167        ) -> BoxFuture<'static, anyhow::Result<SequencerContext<N, P>>>,
168    {
169        // Create a channel to send the context to the web server after it is initialized. This
170        // allows the web server to start before initialization can complete, since initialization
171        // can take a long time (and is dependent on other nodes).
172        let (send_ctx, recv_ctx) = oneshot::channel();
173        let state = ApiState::new(async move {
174            recv_ctx
175                .await
176                .expect("context initialized and sent over channel")
177        });
178        let mut tasks = TaskList::default();
179
180        // The server state type depends on whether we are running a query or status API or not, so
181        // we handle the two cases differently.
182        #[allow(clippy::type_complexity)]
183        let (metrics, consumer, storage): (
184            Box<dyn Metrics>,
185            Box<dyn EventConsumer>,
186            Option<RequestResponseStorage>,
187        ) = if let Some(query_opt) = self.query.take() {
188            if let Some(opt) = self.storage_sql.take() {
189                self.init_with_query_module_sql(query_opt, opt, state, &mut tasks)
190                    .await?
191            } else if let Some(opt) = self.storage_fs.take() {
192                self.init_with_query_module_fs(query_opt, opt, state, &mut tasks)
193                    .await?
194            } else {
195                bail!("query module requested but not storage provided");
196            }
197        } else if self.status.is_some() {
198            // If a status API is requested but no availability API, we use the
199            // `MetricsDataSource`, which allows us to run the status API with no persistent
200            // storage.
201            let ds = MetricsDataSource::default();
202            let metrics = ds.populate_metrics();
203            telemetry::set_registry(Arc::new(ds.metrics().registry().clone()));
204            tasks.spawn("process_metrics", ProcessMetrics::new(ds.metrics()).run());
205            let axum_ds = Arc::new(ExtensibleDataSource::new(ds, state.clone()));
206
207            let port = self.http.port;
208            let env_vars = get_public_env_vars().unwrap_or_default();
209            let node_cfg = self.public_node_config.as_deref().cloned();
210            let modules = espresso_api::OptionalModules {
211                submit: self.submit.is_some(),
212                catchup: self.catchup.is_some(),
213                config: self.config.is_some(),
214                hotshot_events: self.hotshot_events.is_some(),
215                ..Default::default()
216            };
217            let max_connections = self.http.max_connections;
218            tasks.spawn("API server", async move {
219                let state = NodeApiStateImpl::new(axum_ds)
220                    .with_env_vars(env_vars)
221                    .with_public_node_config(node_cfg);
222                if let Err(e) =
223                    espresso_api::serve_axum_status(port, state, modules, max_connections).await
224                {
225                    tracing::error!("Axum server error: {}", e);
226                }
227                anyhow::Ok(())
228            });
229
230            if self.http.tonic_port.is_some() {
231                tracing::warn!("gRPC API not available in status-only mode");
232            }
233
234            (metrics, Box::new(NullEventConsumer), None)
235        } else {
236            // If no status or availability API is requested, we don't need metrics or a query
237            // service data source. The only app state is the HotShot handle, which we use to
238            // submit transactions.
239            //
240            // If we have no availability API, we cannot load a saved leaf from local storage,
241            // so we better have been provided the leaf ahead of time if we want it at all.
242            let port = self.http.port;
243            let env_vars = get_public_env_vars().unwrap_or_default();
244            let node_cfg = self.public_node_config.as_deref().cloned();
245            let modules = espresso_api::OptionalModules {
246                submit: self.submit.is_some(),
247                catchup: self.catchup.is_some(),
248                config: self.config.is_some(),
249                hotshot_events: self.hotshot_events.is_some(),
250                ..Default::default()
251            };
252            let axum_ds = Arc::new(state.clone());
253            let max_connections = self.http.max_connections;
254            tasks.spawn("API server", async move {
255                let state = NodeApiStateImpl::new(axum_ds)
256                    .with_env_vars(env_vars)
257                    .with_public_node_config(node_cfg);
258                if let Err(e) =
259                    espresso_api::serve_axum_bare(port, state, modules, max_connections).await
260                {
261                    tracing::error!("Axum server error: {}", e);
262                }
263                anyhow::Ok(())
264            });
265
266            (Box::new(NoMetrics), Box::new(NullEventConsumer), None)
267        };
268
269        let ctx = init_context(metrics, consumer, storage.clone()).await?;
270        send_ctx
271            .send(ctx.clone())
272            .ok()
273            .context("API server exited without receiving context")?;
274        Ok(ctx.with_task_list(tasks))
275    }
276
277    async fn init_with_query_module_fs<N, P>(
278        &self,
279        query_opt: Query,
280        mod_opt: persistence::fs::Options,
281        state: ApiState<N, P>,
282        tasks: &mut TaskList,
283    ) -> anyhow::Result<(
284        Box<dyn Metrics>,
285        Box<dyn EventConsumer>,
286        Option<RequestResponseStorage>,
287    )>
288    where
289        N: ConnectedNetwork<PubKey>,
290        P: SequencerPersistence,
291    {
292        let ds = <fs::DataSource as SequencerDataSource>::create(
293            mod_opt,
294            provider(
295                query_opt.peers,
296                &state,
297                query_opt.light_client,
298                query_opt.light_client_db,
299            )
300            .await?,
301            false,
302        )
303        .await?;
304
305        // Get the inner storage from the data source
306        let inner_storage = ds.inner();
307
308        tasks.spawn("process_metrics", ProcessMetrics::new(ds.metrics()).run());
309
310        let (metrics, ds) = init_query_data_source(ds, state.clone());
311
312        let port = self.http.port;
313        let ds_for_axum = ds.clone();
314        let env_vars = get_public_env_vars().unwrap_or_default();
315        let node_cfg = self.public_node_config.as_deref().cloned();
316        let modules = espresso_api::OptionalModules {
317            submit: self.submit.is_some(),
318            config: self.config.is_some(),
319            hotshot_events: self.hotshot_events.is_some(),
320            ..Default::default()
321        };
322        let max_connections = self.http.max_connections;
323        tasks.spawn("API server", async move {
324            let state = NodeApiStateImpl::new(ds_for_axum)
325                .with_env_vars(env_vars)
326                .with_public_node_config(node_cfg);
327            if let Err(e) = espresso_api::serve_axum_fs(port, state, modules, max_connections).await
328            {
329                tracing::error!("Axum server error: {}", e);
330            }
331            anyhow::Ok(())
332        });
333
334        if self.http.tonic_port.is_some() {
335            tracing::warn!("gRPC API not available with filesystem storage");
336        }
337
338        Ok((
339            metrics,
340            Box::new(ApiEventConsumer::from(ds)),
341            Some(RequestResponseStorage::Fs(inner_storage)),
342        ))
343    }
344
345    async fn init_with_query_module_sql<N, P>(
346        self,
347        query_opt: Query,
348        mod_opt: persistence::sql::Options,
349        state: ApiState<N, P>,
350        tasks: &mut TaskList,
351    ) -> anyhow::Result<(
352        Box<dyn Metrics>,
353        Box<dyn EventConsumer>,
354        Option<RequestResponseStorage>,
355    )>
356    where
357        N: ConnectedNetwork<PubKey>,
358        P: SequencerPersistence,
359    {
360        let mut provider = Provider::default();
361
362        // Use the database itself as a fetching provider: sometimes we can fetch data that is
363        // missing from the query service from ephemeral consensus storage.
364        let db_provider = mod_opt.clone().create().await?;
365        provider = provider
366            .with_block_provider(db_provider.clone())
367            .with_vid_common_provider(db_provider);
368        // If that fails, fetch missing data from peers.
369        provider = provider.with_provider(
370            LightClientProvider::new(
371                query_opt.peers,
372                state.clone(),
373                query_opt.light_client,
374                query_opt.light_client_db,
375            )
376            .await?,
377        );
378
379        let ds = sql::DataSource::create(mod_opt.clone(), provider, false).await?;
380        let inner_storage = ds.inner();
381        tasks.spawn("process_metrics", ProcessMetrics::new(ds.metrics()).run());
382        let (metrics, ds) = init_query_data_source(ds, state.clone());
383
384        let get_node_state = {
385            let state = state.clone();
386            async move { state.node_state().await.clone() }
387        };
388        tasks.spawn(
389            "merklized state storage update loop",
390            update_state_storage_loop(ds.clone(), get_node_state),
391        );
392
393        let port = self.http.port;
394        let ds_for_axum = ds.clone();
395        let env_vars = get_public_env_vars().unwrap_or_default();
396        let node_cfg = self.public_node_config.as_deref().cloned();
397        let modules = espresso_api::OptionalModules {
398            submit: self.submit.is_some(),
399            config: self.config.is_some(),
400            explorer: self.explorer.is_some(),
401            light_client: self.light_client.is_some(),
402            hotshot_events: self.hotshot_events.is_some(),
403            ..Default::default()
404        };
405        let max_connections = self.http.max_connections;
406        tasks.spawn("API server", async move {
407            let state = NodeApiStateImpl::new(ds_for_axum)
408                .with_env_vars(env_vars)
409                .with_public_node_config(node_cfg);
410            if let Err(e) = espresso_api::serve_axum(port, state, modules, max_connections).await {
411                tracing::error!("Axum server error: {}", e);
412            }
413            anyhow::Ok(())
414        });
415
416        if let Some(tonic_port) = self.http.tonic_port {
417            let ds_for_tonic = ds.clone();
418            tasks.spawn("Tonic gRPC server", async move {
419                let state = NodeApiStateImpl::new(ds_for_tonic);
420                if let Err(e) = espresso_api::serve_tonic(tonic_port, state).await {
421                    tracing::error!("Tonic gRPC server error: {}", e);
422                }
423            });
424        }
425
426        Ok((
427            metrics,
428            Box::new(ApiEventConsumer::from(ds)),
429            Some(RequestResponseStorage::Sql(inner_storage)),
430        ))
431    }
432}
433
434/// The minimal HTTP API.
435///
436/// The API automatically includes health and version endpoints. Additional API modules can be
437/// added by including the query-api or submit-api modules.
438#[derive(Parser, Clone, Copy, Debug)]
439pub struct Http {
440    /// Port that the HTTP API will use.
441    #[clap(long, env = "ESPRESSO_NODE_API_PORT", default_value = "8080")]
442    pub port: u16,
443
444    /// Maximum number of concurrent HTTP connections the server will allow.
445    ///
446    /// Connections exceeding this will receive and immediate 429 response and be closed.
447    ///
448    /// Leave unset for no connection limit.
449    #[clap(long, env = "ESPRESSO_NODE_API_MAX_CONNECTIONS")]
450    pub max_connections: Option<usize>,
451
452    /// Optional port for Tonic gRPC API server.
453    #[clap(long, env = "ESPRESSO_NODE_TONIC_PORT")]
454    pub tonic_port: Option<u16>,
455}
456
457impl Http {
458    /// Default options for running a web server on the given port.
459    pub fn with_port(port: u16) -> Self {
460        Self {
461            port,
462            max_connections: None,
463            tonic_port: None,
464        }
465    }
466}
467
468/// Options for the submission API module.
469#[derive(Parser, Clone, Copy, Debug, Default)]
470pub struct Submit;
471
472/// Options for the status API module.
473#[derive(Parser, Clone, Copy, Debug, Default)]
474pub struct Status;
475
476/// Options for the catchup API module.
477#[derive(Parser, Clone, Copy, Debug, Default)]
478pub struct Catchup;
479
480/// Options for the config API module.
481#[derive(Parser, Clone, Copy, Debug, Default)]
482pub struct Config;
483
484/// Options for the query API module.
485#[derive(Parser, Clone, Debug, Default)]
486pub struct Query {
487    /// Peers for fetching missing data for the query service.
488    #[clap(long, env = "ESPRESSO_NODE_API_PEERS", value_delimiter = ',')]
489    pub peers: Vec<Url>,
490
491    /// Light client configuration, for fetching data from peers.
492    #[clap(flatten)]
493    pub light_client: LightClientOptions,
494
495    /// Persistence for the light client, enabling faster startup.
496    #[clap(flatten)]
497    pub light_client_db: LightClientSqliteOptions,
498}
499
500#[cfg(test)]
501impl Query {
502    pub fn test() -> Self {
503        Self::default()
504    }
505}
506
507/// Options for the state API module.
508#[derive(Parser, Clone, Copy, Debug, Default)]
509pub struct State;
510
511/// Options for the Hotshot events streaming API module.
512#[derive(Parser, Clone, Copy, Debug, Default)]
513pub struct HotshotEvents;
514
515/// Options for the explorer API module.
516#[derive(Parser, Clone, Copy, Debug, Default)]
517pub struct Explorer;
518
519/// Options for the light client API module.
520#[derive(Parser, Clone, Copy, Debug, Default)]
521pub struct LightClient;
522
523/// Populate consensus metrics on `ds`, deposit its prometheus registry for the in-process
524/// telemetry push task (idempotent), and wrap it with the API state.
525///
526/// Returns the metrics handle plus the wrapped query data source shared by the axum server and
527/// update loops.
528#[allow(clippy::type_complexity)]
529fn init_query_data_source<N, P, D>(
530    ds: D,
531    state: ApiState<N, P>,
532) -> (Box<dyn Metrics>, Arc<StorageState<N, P, D>>)
533where
534    N: ConnectedNetwork<PubKey>,
535    P: SequencerPersistence,
536    D: SequencerDataSource + CatchupStorage + PruningDataSource + Send + Sync + 'static,
537{
538    let metrics = ds.populate_metrics();
539    telemetry::set_registry(Arc::new(ds.metrics().registry().clone()));
540    let ds = Arc::new(ExtensibleDataSource::new(ds, state));
541    (metrics, ds)
542}
543
544/// The environment variables listed in `api/public-env-vars.toml`, as `KEY=value` strings.
545fn get_public_env_vars() -> anyhow::Result<Vec<String>> {
546    #[derive(serde::Deserialize)]
547    struct PublicEnvVars {
548        variables: BTreeSet<String>,
549    }
550
551    let PublicEnvVars { variables } =
552        toml::from_str(include_str!("../../api/public-env-vars.toml"))?;
553    Ok(variables
554        .into_iter()
555        .map(|key| {
556            let value = env::var(&key).unwrap_or_default();
557            format!("{key}={value}")
558        })
559        .collect())
560}
espresso_node/api/options.rs

espresso_node/api/
options.rs
