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_api::v1::{
8    database::{MigrationStatus, TableSize},
9    node::StakeTableWithEpochNumber,
10    status::NodePublicKeys,
11};
12use espresso_types::{
13    Certificate2, FeeAccount, FeeAccountProof, FeeMerkleTree, Leaf2, NodeState, PubKey,
14    Transaction,
15    config::PublicNetworkConfig,
16    v0::traits::PersistenceOptions,
17    v0_3::{
18        AuthenticatedValidator, ChainConfig, RegisteredValidator, RewardAccountProofV1,
19        RewardAccountQueryDataV1, RewardAccountV1, RewardAmount, RewardMerkleTreeV1,
20        StakeTableEvent,
21    },
22    v0_4::{RewardAccountProofV2, RewardAccountQueryDataV2, RewardAccountV2, RewardMerkleTreeV2},
23};
24use futures::future::{BoxFuture, Future};
25use hotshot_query_service::{
26    availability::{AvailabilityDataSource, BlockQueryData, LeafQueryData, VidCommonQueryData},
27    data_source::{UpdateDataSource, VersionedDataSource},
28    fetching::provider::AnyProvider,
29    node::NodeDataSource,
30    status::StatusDataSource,
31};
32use hotshot_types::{
33    PeerConfig,
34    data::{EpochNumber, VidShare, ViewNumber},
35    light_client::LCV3StateSignatureRequestBody,
36    signature_key::BLSPubKey,
37    simple_certificate::LightClientStateUpdateCertificateV2,
38    traits::node_implementation::NodeType,
39};
40use indexmap::IndexMap;
41use light_client::{state::LightClientOptions, storage::LightClientDbOptions};
42use url::Url;
43
44use super::{
45    AccountQueryData, BlocksFrontier, fs,
46    options::{Options, Query},
47};
48use crate::{
49    SeqTypes, U256,
50    api::{ApiState, LightClientProvider},
51    persistence,
52    state_cert::StateCertFetchError,
53};
54
55pub trait DataSourceOptions: PersistenceOptions {
56    type DataSource: SequencerDataSource<Options = Self>;
57
58    fn enable_query_module(&self, opt: Options, query: Query) -> Options;
59
60    /// Where a node on this storage keeps its light client database.
61    fn light_client_db(&self, query: &Query) -> anyhow::Result<LightClientDbOptions>;
62}
63
64impl DataSourceOptions for persistence::fs::Options {
65    type DataSource = fs::DataSource;
66
67    fn enable_query_module(&self, opt: Options, query: Query) -> Options {
68        opt.query_fs(query, self.clone())
69    }
70
71    fn light_client_db(&self, query: &Query) -> anyhow::Result<LightClientDbOptions> {
72        Ok(LightClientDbOptions::Sqlite(query.sqlite_light_client_db()))
73    }
74}
75
76/// A data source with sequencer-specific functionality.
77///
78/// This trait extends the generic [`AvailabilityDataSource`] with some additional data needed to
79/// provided sequencer-specific endpoints.
80#[async_trait]
81pub trait SequencerDataSource:
82    AvailabilityDataSource<SeqTypes>
83    + NodeDataSource<SeqTypes>
84    + StatusDataSource
85    + UpdateDataSource<SeqTypes>
86    + VersionedDataSource
87    + Sized
88{
89    type Options: DataSourceOptions<DataSource = Self>;
90
91    /// Instantiate a data source from command line options.
92    async fn create(opt: Self::Options, provider: Provider, reset: bool) -> anyhow::Result<Self>;
93}
94
95/// Provider for fetching missing data for the query service.
96pub type Provider = AnyProvider<SeqTypes>;
97
98/// Create a provider for fetching missing data from a list of peer query services.
99pub(super) async fn provider(
100    peers: impl IntoIterator<Item = Url>,
101    state: &ApiState,
102    opt: LightClientOptions,
103    db_opt: LightClientDbOptions,
104) -> anyhow::Result<Provider> {
105    Ok(Provider::default().with_provider(LightClientProvider::new(
106        peers,
107        state.clone(),
108        opt,
109        db_opt,
110    )?))
111}
112
113pub(crate) trait SubmitDataSource {
114    fn submit(
115        &self,
116        tx: Transaction,
117    ) -> impl Send + Future<Output = anyhow::Result<Commitment<Transaction>>>;
118}
119
120pub(crate) trait HotShotConfigDataSource {
121    fn get_config(&self) -> impl Send + Future<Output = PublicNetworkConfig>;
122}
123
124#[async_trait]
125pub(crate) trait StateSignatureDataSource {
126    async fn get_state_signature(&self, height: u64) -> Option<LCV3StateSignatureRequestBody>;
127}
128
129pub(crate) trait NodeStateDataSource {
130    fn node_state(&self) -> impl Send + Future<Output = NodeState>;
131}
132
133pub(crate) trait NodeKeysDataSource {
134    fn node_public_keys(&self) -> impl Send + Future<Output = Option<NodePublicKeys>>;
135}
136
137pub(crate) trait TokenDataSource<T: NodeType> {
138    fn get_initial_supply_l1(&self) -> impl Send + Future<Output = anyhow::Result<U256>>;
139    fn get_total_supply_l1(&self) -> impl Send + Future<Output = anyhow::Result<U256>>;
140    fn get_decided_header(&self) -> impl Send + Future<Output = espresso_types::Header>;
141}
142
143pub(crate) trait StakeTableDataSource<T: NodeType> {
144    /// Get the stake table for a given epoch
145    fn get_stake_table(
146        &self,
147        epoch: Option<EpochNumber>,
148    ) -> impl Send + Future<Output = anyhow::Result<Vec<PeerConfig<T>>>>;
149
150    /// Get the stake table for the current epoch if not provided
151    fn get_stake_table_current(
152        &self,
153    ) -> impl Send + Future<Output = anyhow::Result<StakeTableWithEpochNumber<T>>>;
154
155    /// Get the DA stake table for a given epoch
156    fn get_da_stake_table(
157        &self,
158        epoch: Option<EpochNumber>,
159    ) -> impl Send + Future<Output = anyhow::Result<Vec<PeerConfig<T>>>>;
160
161    /// Get the DA stake table for the current epoch if not provided
162    fn get_da_stake_table_current(
163        &self,
164    ) -> impl Send + Future<Output = anyhow::Result<StakeTableWithEpochNumber<T>>>;
165
166    /// Get all the validators
167    fn get_validators(
168        &self,
169        epoch: EpochNumber,
170    ) -> impl Send + Future<Output = anyhow::Result<IndexMap<Address, AuthenticatedValidator<BLSPubKey>>>>;
171
172    fn get_block_reward(
173        &self,
174        epoch: Option<EpochNumber>,
175    ) -> impl Send + Future<Output = anyhow::Result<Option<RewardAmount>>>;
176    /// Get the current proposal participation.
177    fn current_proposal_participation(
178        &self,
179    ) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>>;
180
181    /// Get the proposal participation for a given epoch.
182    fn proposal_participation(
183        &self,
184        epoch: EpochNumber,
185    ) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>>;
186    /// Get the current vote participation.
187    fn current_vote_participation(&self) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>>;
188
189    /// Get the vote participation for a given epoch.
190    fn vote_participation(
191        &self,
192        epoch: EpochNumber,
193    ) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>>;
194
195    fn get_all_validators(
196        &self,
197        epoch: EpochNumber,
198        offset: u64,
199        limit: u64,
200    ) -> impl Send + Future<Output = anyhow::Result<Vec<RegisteredValidator<PubKey>>>>;
201
202    /// Get stake table events from L1 blocks `from_l1_block..=to_l1_block`.
203    fn stake_table_events(
204        &self,
205        from_l1_block: u64,
206        to_l1_block: u64,
207    ) -> impl Send + Future<Output = anyhow::Result<Vec<StakeTableEvent>>>;
208}
209
210// Thin wrapper trait to access persistence methods from API handlers
211#[async_trait]
212pub(crate) trait StateCertDataSource {
213    async fn get_state_cert_by_epoch(
214        &self,
215        epoch: u64,
216    ) -> anyhow::Result<Option<LightClientStateUpdateCertificateV2<SeqTypes>>>;
217
218    async fn insert_state_cert(
219        &self,
220        epoch: u64,
221        cert: LightClientStateUpdateCertificateV2<SeqTypes>,
222    ) -> anyhow::Result<()>;
223}
224
225pub(crate) trait CatchupDataSource: Sync {
226    /// Get the state of the requested `account`.
227    ///
228    /// The state is fetched from a snapshot at the given height and view, which _must_ correspond!
229    /// `height` is provided to simplify lookups for backends where data is not indexed by view.
230    /// This function is intended to be used for catchup, so `view` should be no older than the last
231    /// decided view.
232    fn get_account(
233        &self,
234        instance: &NodeState,
235        height: u64,
236        view: ViewNumber,
237        account: FeeAccount,
238    ) -> impl Send + Future<Output = anyhow::Result<AccountQueryData>> {
239        async move {
240            let tree = self
241                .get_accounts(instance, height, view, &[account])
242                .await?;
243            let (proof, balance) = FeeAccountProof::prove(&tree, account.into()).context(
244                format!("account {account} not available for height {height}, view {view}"),
245            )?;
246            Ok(AccountQueryData { balance, proof })
247        }
248    }
249
250    /// Get the state of the requested `accounts`.
251    ///
252    /// The state is fetched from a snapshot at the given height and view, which _must_ correspond!
253    /// `height` is provided to simplify lookups for backends where data is not indexed by view.
254    /// This function is intended to be used for catchup, so `view` should be no older than the last
255    /// decided view.
256    fn get_accounts(
257        &self,
258        instance: &NodeState,
259        height: u64,
260        view: ViewNumber,
261        accounts: &[FeeAccount],
262    ) -> impl Send + Future<Output = anyhow::Result<FeeMerkleTree>>;
263
264    /// Get the blocks Merkle tree frontier.
265    ///
266    /// The state is fetched from a snapshot at the given height and view, which _must_ correspond!
267    /// `height` is provided to simplify lookups for backends where data is not indexed by view.
268    /// This function is intended to be used for catchup, so `view` should be no older than the last
269    /// decided view.
270    fn get_frontier(
271        &self,
272        instance: &NodeState,
273        height: u64,
274        view: ViewNumber,
275    ) -> impl Send + Future<Output = anyhow::Result<BlocksFrontier>>;
276
277    fn get_chain_config(
278        &self,
279        commitment: Commitment<ChainConfig>,
280    ) -> impl Send + Future<Output = anyhow::Result<ChainConfig>>;
281
282    fn get_leaf_chain(
283        &self,
284        height: u64,
285    ) -> impl Send + Future<Output = anyhow::Result<Vec<Leaf2>>>;
286
287    /// Load the cert2 stored at exactly `height`, if one exists.
288    ///
289    /// Returns `None` when no cert2 is stored at exactly `height`.
290    fn get_cert2(
291        &self,
292        _height: u64,
293    ) -> impl Send + Future<Output = anyhow::Result<Option<Certificate2<SeqTypes>>>> {
294        async { Ok(None) }
295    }
296
297    /// Get the state of the requested `account`.
298    ///
299    /// The state is fetched from a snapshot at the given height and view, which _must_ correspond!
300    /// `height` is provided to simplify lookups for backends where data is not indexed by view.
301    /// This function is intended to be used for catchup, so `view` should be no older than the last
302    /// decided view.
303    fn get_reward_account_v2(
304        &self,
305        instance: &NodeState,
306        height: u64,
307        view: ViewNumber,
308        account: RewardAccountV2,
309    ) -> impl Send + Future<Output = anyhow::Result<RewardAccountQueryDataV2>> {
310        async move {
311            let tree = self
312                .get_reward_accounts_v2(instance, height, view, &[account])
313                .await?;
314            let (proof, balance) = RewardAccountProofV2::prove(&tree, account.into()).context(
315                format!("reward account {account} not available for height {height}, view {view}"),
316            )?;
317            Ok(RewardAccountQueryDataV2 { balance, proof })
318        }
319    }
320
321    fn get_reward_accounts_v2(
322        &self,
323        instance: &NodeState,
324        height: u64,
325        view: ViewNumber,
326        accounts: &[RewardAccountV2],
327    ) -> impl Send + Future<Output = anyhow::Result<RewardMerkleTreeV2>>;
328
329    fn get_reward_account_v1(
330        &self,
331        instance: &NodeState,
332        height: u64,
333        view: ViewNumber,
334        account: RewardAccountV1,
335    ) -> impl Send + Future<Output = anyhow::Result<RewardAccountQueryDataV1>> {
336        async move {
337            let tree = self
338                .get_reward_accounts_v1(instance, height, view, &[account])
339                .await?;
340            let (proof, balance) = RewardAccountProofV1::prove(&tree, account.into()).context(
341                format!("reward account {account} not available for height {height}, view {view}"),
342            )?;
343            Ok(RewardAccountQueryDataV1 { balance, proof })
344        }
345    }
346
347    fn get_reward_accounts_v1(
348        &self,
349        instance: &NodeState,
350        height: u64,
351        view: ViewNumber,
352        accounts: &[RewardAccountV1],
353    ) -> impl Send + Future<Output = anyhow::Result<RewardMerkleTreeV1>>;
354
355    fn get_reward_merkle_tree_v2(
356        &self,
357        height: u64,
358        view: ViewNumber,
359    ) -> impl Send + Future<Output = anyhow::Result<Vec<u8>>>;
360
361    fn get_state_cert(
362        &self,
363        epoch: u64,
364    ) -> impl Send + Future<Output = anyhow::Result<LightClientStateUpdateCertificateV2<SeqTypes>>>;
365}
366
367pub trait RequestResponseDataSource<Types: NodeType> {
368    fn request_vid_shares(
369        &self,
370        block_number: u64,
371        vid_common_data: VidCommonQueryData<Types>,
372        duration: Duration,
373    ) -> impl Future<Output = BoxFuture<'static, anyhow::Result<Vec<VidShare>>>> + Send;
374}
375
376#[async_trait]
377pub trait StateCertFetchingDataSource<Types: NodeType> {
378    async fn request_state_cert(
379        &self,
380        epoch: u64,
381        timeout: Duration,
382    ) -> Result<LightClientStateUpdateCertificateV2<Types>, StateCertFetchError>;
383}
384
385/// Data source for database metadata and statistics.
386///
387/// This trait is only implemented by SQL-based storage backends (PostgreSQL and SQLite).
388pub(crate) trait DatabaseMetadataSource {
389    /// Get the sizes of all tables in the database.
390    fn get_table_sizes(&self) -> impl Send + Future<Output = anyhow::Result<Vec<TableSize>>>;
391
392    /// Get the status of all deferred background migrations.
393    fn get_migration_status(
394        &self,
395    ) -> impl Send + Future<Output = anyhow::Result<Vec<MigrationStatus>>>;
396}
397
398// ============================================================================
399// Arc delegation implementations
400// ============================================================================
401// These implementations allow Arc<T> to implement the data source traits
402// when T implements them, which is necessary for NodeApiStateImpl to work
403// with Arc-wrapped data sources.
404
405use std::sync::Arc;
406
407#[async_trait]
408impl<D> StateCertDataSource for Arc<D>
409where
410    D: StateCertDataSource + Sync + Send,
411{
412    async fn get_state_cert_by_epoch(
413        &self,
414        epoch: u64,
415    ) -> anyhow::Result<Option<LightClientStateUpdateCertificateV2<SeqTypes>>> {
416        (*self).get_state_cert_by_epoch(epoch).await
417    }
418
419    async fn insert_state_cert(
420        &self,
421        epoch: u64,
422        cert: LightClientStateUpdateCertificateV2<SeqTypes>,
423    ) -> anyhow::Result<()> {
424        (*self).insert_state_cert(epoch, cert).await
425    }
426}
427
428impl<Types, D> RequestResponseDataSource<Types> for Arc<D>
429where
430    Types: NodeType,
431    D: RequestResponseDataSource<Types> + Send + Sync,
432{
433    async fn request_vid_shares(
434        &self,
435        block_number: u64,
436        vid_common_data: VidCommonQueryData<Types>,
437        timeout_duration: Duration,
438    ) -> BoxFuture<'static, anyhow::Result<Vec<VidShare>>> {
439        self.as_ref()
440            .request_vid_shares(block_number, vid_common_data, timeout_duration)
441            .await
442    }
443}
444
445#[async_trait]
446impl<Types, D> StateCertFetchingDataSource<Types> for Arc<D>
447where
448    Types: NodeType,
449    D: StateCertFetchingDataSource<Types> + Sync + Send,
450{
451    async fn request_state_cert(
452        &self,
453        epoch: u64,
454        timeout: Duration,
455    ) -> Result<LightClientStateUpdateCertificateV2<Types>, StateCertFetchError> {
456        (*self).request_state_cert(epoch, timeout).await
457    }
458}
459
460#[async_trait]
461impl<T, D> StakeTableDataSource<T> for Arc<D>
462where
463    T: NodeType,
464    D: StakeTableDataSource<T> + Sync + Send,
465{
466    fn get_stake_table(
467        &self,
468        epoch: Option<EpochNumber>,
469    ) -> impl Send + Future<Output = anyhow::Result<Vec<PeerConfig<T>>>> {
470        let this = self.clone();
471        async move { (*this).get_stake_table(epoch).await }
472    }
473
474    fn get_stake_table_current(
475        &self,
476    ) -> impl Send + Future<Output = anyhow::Result<StakeTableWithEpochNumber<T>>> {
477        let this = self.clone();
478        async move { (*this).get_stake_table_current().await }
479    }
480
481    fn get_da_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_da_stake_table(epoch).await }
487    }
488
489    fn get_da_stake_table_current(
490        &self,
491    ) -> impl Send + Future<Output = anyhow::Result<StakeTableWithEpochNumber<T>>> {
492        let this = self.clone();
493        async move { (*this).get_da_stake_table_current().await }
494    }
495
496    fn get_validators(
497        &self,
498        epoch: EpochNumber,
499    ) -> impl Send + Future<Output = anyhow::Result<IndexMap<Address, AuthenticatedValidator<BLSPubKey>>>>
500    {
501        let this = self.clone();
502        async move { (*this).get_validators(epoch).await }
503    }
504
505    fn get_block_reward(
506        &self,
507        epoch: Option<EpochNumber>,
508    ) -> impl Send + Future<Output = anyhow::Result<Option<RewardAmount>>> {
509        let this = self.clone();
510        async move { (*this).get_block_reward(epoch).await }
511    }
512
513    fn current_proposal_participation(
514        &self,
515    ) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>> {
516        let this = self.clone();
517        async move { (*this).current_proposal_participation().await }
518    }
519
520    fn proposal_participation(
521        &self,
522        epoch: EpochNumber,
523    ) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>> {
524        let this = self.clone();
525        async move { (*this).proposal_participation(epoch).await }
526    }
527
528    fn current_vote_participation(&self) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>> {
529        let this = self.clone();
530        async move { (*this).current_vote_participation().await }
531    }
532
533    fn vote_participation(
534        &self,
535        epoch: EpochNumber,
536    ) -> impl Send + Future<Output = HashMap<BLSPubKey, f64>> {
537        let this = self.clone();
538        async move { (*this).vote_participation(epoch).await }
539    }
540
541    fn get_all_validators(
542        &self,
543        epoch: EpochNumber,
544        offset: u64,
545        limit: u64,
546    ) -> impl Send + Future<Output = anyhow::Result<Vec<RegisteredValidator<PubKey>>>> {
547        let this = self.clone();
548        async move { (*this).get_all_validators(epoch, offset, limit).await }
549    }
550
551    fn stake_table_events(
552        &self,
553        from_l1_block: u64,
554        to_l1_block: u64,
555    ) -> impl Send + Future<Output = anyhow::Result<Vec<StakeTableEvent>>> {
556        let this = self.clone();
557        async move { (*this).stake_table_events(from_l1_block, to_l1_block).await }
558    }
559}
560
561/// Data source for pruning state: the oldest retained block and leaf.
562///
563/// SQL backends return the actual oldest entry; the filesystem backend always returns `None`
564/// since it does not prune.
565pub(crate) trait PruningDataSource {
566    /// Get the oldest block in storage, or `None` if empty or unsupported.
567    fn get_oldest_block(
568        &self,
569    ) -> impl Send + Future<Output = anyhow::Result<Option<BlockQueryData<SeqTypes>>>>;
570
571    /// Get the oldest leaf in storage, or `None` if empty or unsupported.
572    fn get_oldest_leaf(
573        &self,
574    ) -> impl Send + Future<Output = anyhow::Result<Option<LeafQueryData<SeqTypes>>>>;
575}
576
577#[cfg(test)]
578mod test {
579    use hotshot_types::{
580        addr::NetAddr, light_client::StateKeyPair, traits::signature_key::SignatureKey as _, x25519,
581    };
582
583    use super::*;
584
585    #[test]
586    fn test_node_public_keys_serialize_like_stake_table() {
587        let account = Address::random();
588        let consensus_key = BLSPubKey::generated_from_seed_indexed([1; 32], 0).0;
589        let state_ver_key = StateKeyPair::generate_from_seed_indexed([2; 32], 0).ver_key();
590        let x25519_key = x25519::Keypair::generated_from_seed_indexed([3; 32], 0)
591            .unwrap()
592            .public_key();
593        let p2p_addr: NetAddr = "node.example.com:9977".parse().unwrap();
594
595        let validator = serde_json::to_value(RegisteredValidator::<BLSPubKey> {
596            account,
597            stake_table_key: Some(consensus_key),
598            state_ver_key: Some(state_ver_key.clone()),
599            stake: U256::from(1u64),
600            commission: 0,
601            delegators: HashMap::new(),
602            authenticated: true,
603            x25519_key: Some(x25519_key),
604            p2p_addr: Some(p2p_addr.clone()),
605        })
606        .unwrap();
607
608        let keys = serde_json::to_value(NodePublicKeys {
609            eth_account: Some(account),
610            consensus_key,
611            state_ver_key,
612            x25519_key: Some(x25519_key),
613            p2p_addr: Some(p2p_addr),
614        })
615        .unwrap();
616
617        assert_eq!(keys["eth_account"], validator["account"]);
618        assert_eq!(keys["consensus_key"], validator["stake_table_key"]);
619        assert_eq!(keys["state_ver_key"], validator["state_ver_key"]);
620        assert_eq!(keys["x25519_key"], x25519_key.to_string());
621        assert_eq!(keys["p2p_addr"], validator["p2p_addr"]);
622        assert_eq!(keys["p2p_addr"], "node.example.com:9977");
623    }
624
625    #[test]
626    fn sql_storage_keeps_light_client_in_postgres() {
627        use clap::Parser as _;
628
629        let mut opt = persistence::sql::Options::try_parse_from([
630            "espresso-node",
631            "--host",
632            "db.example",
633            "--port",
634            "6543",
635            "--database",
636            "espresso",
637        ])
638        .unwrap();
639        // The parsed default follows ESPRESSO_NODE_EMBEDDED_DB, not this test.
640        opt.embedded_db = false;
641        let opt = persistence::sql::postgres::Options::try_from(opt).unwrap();
642        let LightClientDbOptions::Postgres(lc) = opt.light_client_db(&Query::default()).unwrap()
643        else {
644            panic!("expected the light client in Postgres");
645        };
646        assert_eq!(lc.connect.get_host(), "db.example");
647        assert_eq!(lc.connect.get_port(), 6543);
648        assert_eq!(lc.connect.get_database(), Some("espresso"));
649    }
650}
651
652#[cfg(any(test, feature = "testing"))]
653pub mod testing {
654    use espresso_types::BlockMerkleTree;
655    use hotshot_query_service::{
656        data_source::storage::{
657            AvailabilityStorage, NodeStorage, UpdateAvailabilityStorage, sql::StorageConnectionType,
658        },
659        merklized_state::{
660            MerklizedStateDataSource, MerklizedStateHeightPersistence, UpdateStateData,
661        },
662    };
663    use jf_merkle_tree_compat::MerkleTreeScheme as _;
664
665    use super::{super::Options, *};
666    use crate::persistence::sql::Options as SqlOptions;
667
668    #[async_trait]
669    pub trait TestableSequencerDataSource: SequencerDataSource {
670        type Storage: Sync;
671
672        async fn create_storage() -> Self::Storage;
673        fn persistence_options(storage: &Self::Storage) -> Self::Options;
674        fn leaf_only_ds_options(
675            _storage: &Self::Storage,
676            _opt: Options,
677        ) -> anyhow::Result<Options> {
678            anyhow::bail!("not supported")
679        }
680        fn options(storage: &Self::Storage, opt: Options) -> Options;
681    }
682
683    /// A [`TestableSequencerDataSource`] on a SQL database, with the direct access to its rows
684    /// that some tests need to set up or inspect a node's state.
685    #[async_trait]
686    pub trait TestableSqlDataSource:
687        TestableSequencerDataSource<Storage: Send>
688        + Clone
689        + Send
690        + Sync
691        + 'static
692        + MerklizedStateDataSource<SeqTypes, BlockMerkleTree, { BlockMerkleTree::ARITY }>
693        + MerklizedStateHeightPersistence
694        + for<'a> VersionedDataSource<
695            ReadOnly<'a>: AvailabilityStorage<SeqTypes> + NodeStorage<SeqTypes>,
696            Transaction<'a>: UpdateAvailabilityStorage<SeqTypes>
697                                 + UpdateStateData<
698                SeqTypes,
699                BlockMerkleTree,
700                { BlockMerkleTree::ARITY },
701            >,
702        >
703    {
704        /// A connection to the database, outside any node.
705        type SqlStorage: Send + Sync;
706
707        fn tmp_options(storage: &Self::Storage) -> SqlOptions;
708        fn sql_persistence_options(opt: SqlOptions) -> Self::Options;
709        async fn connect(
710            opt: &SqlOptions,
711            connection_type: StorageConnectionType,
712        ) -> anyhow::Result<Self::SqlStorage>;
713
714        /// A data source on `storage` with `provider` that fetches only what is asked for, with
715        /// neither proactive fetching nor the aggregator running.
716        async fn on_request_data_source(
717            storage: &Self::Storage,
718            provider: super::Provider,
719        ) -> anyhow::Result<Self>;
720
721        async fn prune_state_below(
722            ds: &Self,
723            height: u64,
724            cfg: &hotshot_query_service::data_source::storage::pruning::PrunerCfg,
725        ) -> anyhow::Result<()>;
726
727        /// The number of blocks in availability storage.
728        async fn local_block_height(db: &Self::SqlStorage) -> anyhow::Result<u64>;
729
730        /// The state pruned height, then the state head.
731        async fn state_heights(db: &Self::SqlStorage) -> anyhow::Result<(Option<u64>, u64)>;
732
733        /// The height the state loop recorded as its start, if it started from a boundary.
734        async fn state_start_height(db: &Self::SqlStorage) -> anyhow::Result<Option<u64>>;
735
736        /// The lowest height with a block tree node, which is the first snapshot the state loop
737        /// stored.
738        async fn lowest_block_tree_node(db: &Self::SqlStorage) -> anyhow::Result<u64>;
739
740        async fn insert_leaf(
741            db: &Self::SqlStorage,
742            leaf: &LeafQueryData<SeqTypes>,
743        ) -> anyhow::Result<()>;
744
745        async fn set_last_state_height(db: &Self::SqlStorage, height: u64) -> anyhow::Result<()>;
746
747        /// Stamp the state pruned height without pruning anything.
748        async fn stamp_state_pruned_height(
749            db: &Self::SqlStorage,
750            height: u64,
751        ) -> anyhow::Result<()>;
752
753        /// Prune consensus data to `pruned` as a pruner batch would, stamping the data pruned
754        /// height and then deleting the height-keyed rows `delete_batch` deletes.
755        async fn prune_data_to(db: &Self::SqlStorage, pruned: u64) -> anyhow::Result<()>;
756
757        /// Delete consensus data above `from` up to and including `to` without moving the data
758        /// pruned height, as if the node had never had it.
759        async fn delete_data_between(
760            db: &Self::SqlStorage,
761            from: u64,
762            to: u64,
763        ) -> anyhow::Result<()>;
764
765        /// The number of stored leaves at or below `height`.
766        async fn count_leaves_up_to(db: &Self::SqlStorage, height: u64) -> anyhow::Result<i64>;
767
768        /// The data pruned height, then the lowest stored leaf above genesis.
769        async fn data_heights(db: &Self::SqlStorage) -> anyhow::Result<(Option<u64>, Option<u64>)>;
770
771        /// Record a deferred migration, as the migration runner does when it starts one.
772        async fn insert_deferred_migration(
773            db: &Self::SqlStorage,
774            migration: DeferredMigration,
775        ) -> anyhow::Result<()>;
776
777        async fn reconstruct_state(
778            db: &Self::SqlStorage,
779            instance: &NodeState,
780            from_height: u64,
781            to_view: ViewNumber,
782            fee_accounts: &[FeeAccount],
783            reward_accounts: &[RewardAccountV2],
784        ) -> anyhow::Result<(espresso_types::ValidatedState, Leaf2)>;
785    }
786
787    /// A row of the `deferred_migrations` table.
788    pub struct DeferredMigration {
789        pub name: &'static str,
790        pub started_at: chrono::DateTime<chrono::Utc>,
791        pub completed_at: Option<chrono::DateTime<chrono::Utc>>,
792        pub last_offset: Option<i64>,
793    }
794}
espresso_node/api/data_source.rs

espresso_node/api/
data_source.rs
