espresso_node/api/
options.rs

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

espresso_node/api/
options.rs
