espresso_node/api/
data_source.rs

1use std::{collections::HashMap, time::Duration};
2
3use alloy::primitives::Address;
4use anyhow::Context;
5use async_trait::async_trait;
6use committable::Commitment;
7use espresso_types::{
8    Certificate2, FeeAccount, FeeAccountProof, FeeMerkleTree, Leaf2, NodeState, PubKey,
9    Transaction,
10    config::PublicNetworkConfig,
11    v0::traits::{PersistenceOptions, SequencerPersistence},
12    v0_3::{
13        AuthenticatedValidator, ChainConfig, RegisteredValidator, RewardAccountProofV1,
14        RewardAccountQueryDataV1, RewardAccountV1, RewardAmount, RewardMerkleTreeV1,
15        StakeTableEvent,
16    },
17    v0_4::{RewardAccountProofV2, RewardAccountQueryDataV2, RewardAccountV2, RewardMerkleTreeV2},
18};
19use futures::future::{BoxFuture, Future};
20use hotshot::types::BLSPubKey;
21use hotshot_query_service::{
22    availability::{AvailabilityDataSource, BlockQueryData, LeafQueryData, VidCommonQueryData},
23    data_source::{UpdateDataSource, VersionedDataSource},
24    fetching::provider::AnyProvider,
25    node::NodeDataSource,
26    status::StatusDataSource,
27};
28use hotshot_types::{
29    PeerConfig,
30    data::{EpochNumber, VidShare, ViewNumber},
31    light_client::LCV3StateSignatureRequestBody,
32    simple_certificate::LightClientStateUpdateCertificateV2,
33    traits::{network::ConnectedNetwork, node_implementation::NodeType},
34};
35use indexmap::IndexMap;
36use light_client::{state::LightClientOptions, storage::LightClientSqliteOptions};
37use serde::{Deserialize, Serialize};
38use tide_disco::Url;
39
40use super::{
41    AccountQueryData, BlocksFrontier, fs,
42    options::{Options, Query},
43    sql,
44};
45use crate::{
46    SeqTypes, U256,
47    api::{ApiState, LightClientProvider},
48    persistence,
49    state_cert::StateCertFetchError,
50};
51
52pub trait DataSourceOptions: PersistenceOptions {
53    type DataSource: SequencerDataSource<Options = Self>;
54
55    fn enable_query_module(&self, opt: Options, query: Query) -> Options;
56}
57
58impl DataSourceOptions for persistence::sql::Options {
59    type DataSource = sql::DataSource;
60
61    fn enable_query_module(&self, opt: Options, query: Query) -> Options {
62        opt.query_sql(query, self.clone())
63    }
64}
65
66impl DataSourceOptions for persistence::fs::Options {
67    type DataSource = fs::DataSource;
68
69    fn enable_query_module(&self, opt: Options, query: Query) -> Options {
70        opt.query_fs(query, self.clone())
71    }
72}
73
74/// A data source with sequencer-specific functionality.
75///
76/// This trait extends the generic [`AvailabilityDataSource`] with some additional data needed to
77/// provided sequencer-specific endpoints.
78#[async_trait]
79pub trait SequencerDataSource:
80    AvailabilityDataSource<SeqTypes>
81    + NodeDataSource<SeqTypes>
82    + StatusDataSource
83    + UpdateDataSource<SeqTypes>
84    + VersionedDataSource
85    + Sized
86{
87    type Options: DataSourceOptions<DataSource = Self>;
88
89    /// Instantiate a data source from command line options.
90    async fn create(opt: Self::Options, provider: Provider, reset: bool) -> anyhow::Result<Self>;
91}
92
93/// Provider for fetching missing data for the query service.
94pub type Provider = AnyProvider<SeqTypes>;
95
96/// Create a provider for fetching missing data from a list of peer query services.
97pub(super) async fn provider<N, P>(
98    peers: impl IntoIterator<Item = Url>,
99    state: &ApiState<N, P>,
100    opt: LightClientOptions,
101    db_opt: LightClientSqliteOptions,
102) -> anyhow::Result<Provider>
103where
104    N: ConnectedNetwork<PubKey>,
105    P: SequencerPersistence,
106{
107    Ok(Provider::default()
108        .with_provider(LightClientProvider::new(peers, state.clone(), opt, db_opt).await?))
109}
110
111pub(crate) trait SubmitDataSource<N: ConnectedNetwork<PubKey>, P: SequencerPersistence> {
112    fn submit(&self, tx: Transaction) -> impl Send + Future<Output = anyhow::Result<()>>;
113}
114
115pub(crate) trait HotShotConfigDataSource {
116    fn get_config(&self) -> impl Send + Future<Output = PublicNetworkConfig>;
117}
118
119#[async_trait]
120pub(crate) trait StateSignatureDataSource<N: ConnectedNetwork<PubKey>> {
121    async fn get_state_signature(&self, height: u64) -> Option<LCV3StateSignatureRequestBody>;
122}
123
124pub(crate) trait NodeStateDataSource {
125    fn node_state(&self) -> impl Send + Future<Output = NodeState>;
126}
127
128pub(crate) trait TokenDataSource<T: NodeType> {
129    fn get_initial_supply_l1(&self) -> impl Send + Future<Output = anyhow::Result<U256>>;
130    fn get_total_supply_l1(&self) -> impl Send + Future<Output = anyhow::Result<U256>>;
131    fn get_decided_header(&self) -> impl Send + Future<Output = espresso_types::Header>;
132}
133
134#[derive(Serialize, Deserialize)]
135#[serde(bound = "T: NodeType")]
136pub struct StakeTableWithEpochNumber<T: NodeType> {
137    pub epoch: Option<EpochNumber>,
138    pub stake_table: Vec<PeerConfig<T>>,
139}
140
141pub(crate) trait StakeTableDataSource<T: NodeType> {
142    /// Get the stake table for a given epoch
143    fn get_stake_table(
144        &self,
145        epoch: Option<EpochNumber>,
146    ) -> impl Send + Future<Output = anyhow::Result<Vec<PeerConfig<T>>>>;
147
148    /// Get the stake table for the current epoch if not provided
149    fn get_stake_table_current(
150        &self,
151    ) -> impl Send + Future<Output = anyhow::Result<StakeTableWithEpochNumber<T>>>;
152
153    /// Get the DA stake table for a given epoch
154    fn get_da_stake_table(
155        &self,
156        epoch: Option<EpochNumber>,
157    ) -> impl Send + Future<Output = anyhow::Result<Vec<PeerConfig<T>>>>;
158
159    /// Get the DA stake table for the current epoch if not provided
160    fn get_da_stake_table_current(
161        &self,
162    ) -> impl Send + Future<Output = anyhow::Result<StakeTableWithEpochNumber<T>>>;
163
164    /// Get all the validators
165    fn get_validators(
166        &self,
167        epoch: EpochNumber,
168    ) -> impl Send + Future<Output = anyhow::Result<IndexMap<Address, AuthenticatedValidator<BLSPubKey>>>>;
169
170    fn get_block_reward(
171        &self,
172        epoch: Option<EpochNumber>,
173    ) -> impl Send + Future<Output = anyhow::Result<Option<RewardAmount>>>;
174    /// Get the current proposal participation.
175    fn current_proposal_participation(
176        &self,
177    ) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>>;
178
179    /// Get the proposal participation for a given epoch.
180    fn proposal_participation(
181        &self,
182        epoch: EpochNumber,
183    ) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>>;
184    /// Get the current vote participation.
185    fn current_vote_participation(&self) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>>;
186
187    /// Get the vote participation for a given epoch.
188    fn vote_participation(
189        &self,
190        epoch: EpochNumber,
191    ) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>>;
192
193    fn get_all_validators(
194        &self,
195        epoch: EpochNumber,
196        offset: u64,
197        limit: u64,
198    ) -> impl Send + Future<Output = anyhow::Result<Vec<RegisteredValidator<PubKey>>>>;
199
200    /// Get stake table events from L1 blocks `from_l1_block..=to_l1_block`.
201    fn stake_table_events(
202        &self,
203        from_l1_block: u64,
204        to_l1_block: u64,
205    ) -> impl Send + Future<Output = anyhow::Result<Vec<StakeTableEvent>>>;
206}
207
208// Thin wrapper trait to access persistence methods from API handlers
209#[async_trait]
210pub(crate) trait StateCertDataSource {
211    async fn get_state_cert_by_epoch(
212        &self,
213        epoch: u64,
214    ) -> anyhow::Result<Option<LightClientStateUpdateCertificateV2<SeqTypes>>>;
215
216    async fn insert_state_cert(
217        &self,
218        epoch: u64,
219        cert: LightClientStateUpdateCertificateV2<SeqTypes>,
220    ) -> anyhow::Result<()>;
221}
222
223pub(crate) trait CatchupDataSource: Sync {
224    /// Get the state of the requested `account`.
225    ///
226    /// The state is fetched from a snapshot at the given height and view, which _must_ correspond!
227    /// `height` is provided to simplify lookups for backends where data is not indexed by view.
228    /// This function is intended to be used for catchup, so `view` should be no older than the last
229    /// decided view.
230    fn get_account(
231        &self,
232        instance: &NodeState,
233        height: u64,
234        view: ViewNumber,
235        account: FeeAccount,
236    ) -> impl Send + Future<Output = anyhow::Result<AccountQueryData>> {
237        async move {
238            let tree = self
239                .get_accounts(instance, height, view, &[account])
240                .await?;
241            let (proof, balance) = FeeAccountProof::prove(&tree, account.into()).context(
242                format!("account {account} not available for height {height}, view {view}"),
243            )?;
244            Ok(AccountQueryData { balance, proof })
245        }
246    }
247
248    /// Get the state of the requested `accounts`.
249    ///
250    /// The state is fetched from a snapshot at the given height and view, which _must_ correspond!
251    /// `height` is provided to simplify lookups for backends where data is not indexed by view.
252    /// This function is intended to be used for catchup, so `view` should be no older than the last
253    /// decided view.
254    fn get_accounts(
255        &self,
256        instance: &NodeState,
257        height: u64,
258        view: ViewNumber,
259        accounts: &[FeeAccount],
260    ) -> impl Send + Future<Output = anyhow::Result<FeeMerkleTree>>;
261
262    /// Get the blocks Merkle tree frontier.
263    ///
264    /// The state is fetched from a snapshot at the given height and view, which _must_ correspond!
265    /// `height` is provided to simplify lookups for backends where data is not indexed by view.
266    /// This function is intended to be used for catchup, so `view` should be no older than the last
267    /// decided view.
268    fn get_frontier(
269        &self,
270        instance: &NodeState,
271        height: u64,
272        view: ViewNumber,
273    ) -> impl Send + Future<Output = anyhow::Result<BlocksFrontier>>;
274
275    fn get_chain_config(
276        &self,
277        commitment: Commitment<ChainConfig>,
278    ) -> impl Send + Future<Output = anyhow::Result<ChainConfig>>;
279
280    fn get_leaf_chain(
281        &self,
282        height: u64,
283    ) -> impl Send + Future<Output = anyhow::Result<Vec<Leaf2>>>;
284
285    /// Load the cert2 stored at exactly `height`, if one exists.
286    ///
287    /// Returns `None` when no cert2 is stored at exactly `height`.
288    fn get_cert2(
289        &self,
290        _height: u64,
291    ) -> impl Send + Future<Output = anyhow::Result<Option<Certificate2<SeqTypes>>>> {
292        async { Ok(None) }
293    }
294
295    /// Get the state of the requested `account`.
296    ///
297    /// The state is fetched from a snapshot at the given height and view, which _must_ correspond!
298    /// `height` is provided to simplify lookups for backends where data is not indexed by view.
299    /// This function is intended to be used for catchup, so `view` should be no older than the last
300    /// decided view.
301    fn get_reward_account_v2(
302        &self,
303        instance: &NodeState,
304        height: u64,
305        view: ViewNumber,
306        account: RewardAccountV2,
307    ) -> impl Send + Future<Output = anyhow::Result<RewardAccountQueryDataV2>> {
308        async move {
309            let tree = self
310                .get_reward_accounts_v2(instance, height, view, &[account])
311                .await?;
312            let (proof, balance) = RewardAccountProofV2::prove(&tree, account.into()).context(
313                format!("reward account {account} not available for height {height}, view {view}"),
314            )?;
315            Ok(RewardAccountQueryDataV2 { balance, proof })
316        }
317    }
318
319    fn get_reward_accounts_v2(
320        &self,
321        instance: &NodeState,
322        height: u64,
323        view: ViewNumber,
324        accounts: &[RewardAccountV2],
325    ) -> impl Send + Future<Output = anyhow::Result<RewardMerkleTreeV2>>;
326
327    fn get_reward_account_v1(
328        &self,
329        instance: &NodeState,
330        height: u64,
331        view: ViewNumber,
332        account: RewardAccountV1,
333    ) -> impl Send + Future<Output = anyhow::Result<RewardAccountQueryDataV1>> {
334        async move {
335            let tree = self
336                .get_reward_accounts_v1(instance, height, view, &[account])
337                .await?;
338            let (proof, balance) = RewardAccountProofV1::prove(&tree, account.into()).context(
339                format!("reward account {account} not available for height {height}, view {view}"),
340            )?;
341            Ok(RewardAccountQueryDataV1 { balance, proof })
342        }
343    }
344
345    fn get_reward_accounts_v1(
346        &self,
347        instance: &NodeState,
348        height: u64,
349        view: ViewNumber,
350        accounts: &[RewardAccountV1],
351    ) -> impl Send + Future<Output = anyhow::Result<RewardMerkleTreeV1>>;
352
353    fn get_reward_merkle_tree_v2(
354        &self,
355        height: u64,
356        view: ViewNumber,
357    ) -> impl Send + Future<Output = anyhow::Result<Vec<u8>>>;
358
359    fn get_state_cert(
360        &self,
361        epoch: u64,
362    ) -> impl Send + Future<Output = anyhow::Result<LightClientStateUpdateCertificateV2<SeqTypes>>>;
363}
364
365pub trait RequestResponseDataSource<Types: NodeType> {
366    fn request_vid_shares(
367        &self,
368        block_number: u64,
369        vid_common_data: VidCommonQueryData<Types>,
370        duration: Duration,
371    ) -> impl Future<Output = BoxFuture<'static, anyhow::Result<Vec<VidShare>>>> + Send;
372}
373
374#[async_trait]
375pub trait StateCertFetchingDataSource<Types: NodeType> {
376    async fn request_state_cert(
377        &self,
378        epoch: u64,
379        timeout: Duration,
380    ) -> Result<LightClientStateUpdateCertificateV2<Types>, StateCertFetchError>;
381}
382
383/// Database table size information.
384#[derive(Serialize, Deserialize, Clone, Debug)]
385pub struct TableSize {
386    pub table_name: String,
387    pub row_count: i64,
388    pub total_size_bytes: Option<i64>,
389}
390
391/// Status of a single deferred background migration.
392#[derive(Serialize, Deserialize, Clone, Debug)]
393pub struct MigrationStatus {
394    pub name: String,
395    pub started_at: chrono::DateTime<chrono::Utc>,
396    pub completed_at: Option<chrono::DateTime<chrono::Utc>>,
397    pub last_offset: Option<i64>,
398}
399
400/// Data source for database metadata and statistics.
401///
402/// This trait is only implemented by SQL-based storage backends (PostgreSQL and SQLite).
403pub(crate) trait DatabaseMetadataSource {
404    /// Get the sizes of all tables in the database.
405    fn get_table_sizes(&self) -> impl Send + Future<Output = anyhow::Result<Vec<TableSize>>>;
406
407    /// Get the status of all deferred background migrations.
408    fn get_migration_status(
409        &self,
410    ) -> impl Send + Future<Output = anyhow::Result<Vec<MigrationStatus>>>;
411}
412
413// ============================================================================
414// Arc delegation implementations
415// ============================================================================
416// These implementations allow Arc<T> to implement the data source traits
417// when T implements them, which is necessary for NodeApiStateImpl to work
418// with Arc-wrapped data sources.
419
420use std::sync::Arc;
421
422#[async_trait]
423impl<D> StateCertDataSource for Arc<D>
424where
425    D: StateCertDataSource + Sync + Send,
426{
427    async fn get_state_cert_by_epoch(
428        &self,
429        epoch: u64,
430    ) -> anyhow::Result<Option<LightClientStateUpdateCertificateV2<SeqTypes>>> {
431        (*self).get_state_cert_by_epoch(epoch).await
432    }
433
434    async fn insert_state_cert(
435        &self,
436        epoch: u64,
437        cert: LightClientStateUpdateCertificateV2<SeqTypes>,
438    ) -> anyhow::Result<()> {
439        (*self).insert_state_cert(epoch, cert).await
440    }
441}
442
443impl<Types, D> RequestResponseDataSource<Types> for Arc<D>
444where
445    Types: NodeType,
446    D: RequestResponseDataSource<Types> + Send + Sync,
447{
448    async fn request_vid_shares(
449        &self,
450        block_number: u64,
451        vid_common_data: VidCommonQueryData<Types>,
452        timeout_duration: Duration,
453    ) -> BoxFuture<'static, anyhow::Result<Vec<VidShare>>> {
454        self.as_ref()
455            .request_vid_shares(block_number, vid_common_data, timeout_duration)
456            .await
457    }
458}
459
460#[async_trait]
461impl<Types, D> StateCertFetchingDataSource<Types> for Arc<D>
462where
463    Types: NodeType,
464    D: StateCertFetchingDataSource<Types> + Sync + Send,
465{
466    async fn request_state_cert(
467        &self,
468        epoch: u64,
469        timeout: Duration,
470    ) -> Result<LightClientStateUpdateCertificateV2<Types>, StateCertFetchError> {
471        (*self).request_state_cert(epoch, timeout).await
472    }
473}
474
475#[async_trait]
476impl<T, D> StakeTableDataSource<T> for Arc<D>
477where
478    T: NodeType,
479    D: StakeTableDataSource<T> + Sync + Send,
480{
481    fn get_stake_table(
482        &self,
483        epoch: Option<EpochNumber>,
484    ) -> impl Send + Future<Output = anyhow::Result<Vec<PeerConfig<T>>>> {
485        let this = self.clone();
486        async move { (*this).get_stake_table(epoch).await }
487    }
488
489    fn get_stake_table_current(
490        &self,
491    ) -> impl Send + Future<Output = anyhow::Result<StakeTableWithEpochNumber<T>>> {
492        let this = self.clone();
493        async move { (*this).get_stake_table_current().await }
494    }
495
496    fn get_da_stake_table(
497        &self,
498        epoch: Option<EpochNumber>,
499    ) -> impl Send + Future<Output = anyhow::Result<Vec<PeerConfig<T>>>> {
500        let this = self.clone();
501        async move { (*this).get_da_stake_table(epoch).await }
502    }
503
504    fn get_da_stake_table_current(
505        &self,
506    ) -> impl Send + Future<Output = anyhow::Result<StakeTableWithEpochNumber<T>>> {
507        let this = self.clone();
508        async move { (*this).get_da_stake_table_current().await }
509    }
510
511    fn get_validators(
512        &self,
513        epoch: EpochNumber,
514    ) -> impl Send + Future<Output = anyhow::Result<IndexMap<Address, AuthenticatedValidator<BLSPubKey>>>>
515    {
516        let this = self.clone();
517        async move { (*this).get_validators(epoch).await }
518    }
519
520    fn get_block_reward(
521        &self,
522        epoch: Option<EpochNumber>,
523    ) -> impl Send + Future<Output = anyhow::Result<Option<RewardAmount>>> {
524        let this = self.clone();
525        async move { (*this).get_block_reward(epoch).await }
526    }
527
528    fn current_proposal_participation(
529        &self,
530    ) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>> {
531        let this = self.clone();
532        async move { (*this).current_proposal_participation().await }
533    }
534
535    fn proposal_participation(
536        &self,
537        epoch: EpochNumber,
538    ) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>> {
539        let this = self.clone();
540        async move { (*this).proposal_participation(epoch).await }
541    }
542
543    fn current_vote_participation(&self) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>> {
544        let this = self.clone();
545        async move { (*this).current_vote_participation().await }
546    }
547
548    fn vote_participation(
549        &self,
550        epoch: EpochNumber,
551    ) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>> {
552        let this = self.clone();
553        async move { (*this).vote_participation(epoch).await }
554    }
555
556    fn get_all_validators(
557        &self,
558        epoch: EpochNumber,
559        offset: u64,
560        limit: u64,
561    ) -> impl Send + Future<Output = anyhow::Result<Vec<RegisteredValidator<PubKey>>>> {
562        let this = self.clone();
563        async move { (*this).get_all_validators(epoch, offset, limit).await }
564    }
565
566    fn stake_table_events(
567        &self,
568        from_l1_block: u64,
569        to_l1_block: u64,
570    ) -> impl Send + Future<Output = anyhow::Result<Vec<StakeTableEvent>>> {
571        let this = self.clone();
572        async move { (*this).stake_table_events(from_l1_block, to_l1_block).await }
573    }
574}
575
576/// Data source for pruning state: the oldest retained block and leaf.
577///
578/// SQL backends return the actual oldest entry; the filesystem backend always returns `None`
579/// since it does not prune.
580pub(crate) trait PruningDataSource {
581    /// Get the oldest block in storage, or `None` if empty or unsupported.
582    fn get_oldest_block(
583        &self,
584    ) -> impl Send + Future<Output = anyhow::Result<Option<BlockQueryData<SeqTypes>>>>;
585
586    /// Get the oldest leaf in storage, or `None` if empty or unsupported.
587    fn get_oldest_leaf(
588        &self,
589    ) -> impl Send + Future<Output = anyhow::Result<Option<LeafQueryData<SeqTypes>>>>;
590}
591
592#[cfg(any(test, feature = "testing"))]
593pub mod testing {
594    use super::{super::Options, *};
595
596    #[async_trait]
597    pub trait TestableSequencerDataSource: SequencerDataSource {
598        type Storage: Sync;
599
600        async fn create_storage() -> Self::Storage;
601        fn persistence_options(storage: &Self::Storage) -> Self::Options;
602        fn leaf_only_ds_options(
603            _storage: &Self::Storage,
604            _opt: Options,
605        ) -> anyhow::Result<Options> {
606            anyhow::bail!("not supported")
607        }
608        fn options(storage: &Self::Storage, opt: Options) -> Options;
609    }
610}
espresso_node/api/data_source.rs

espresso_node/api/
data_source.rs
