mithril_signer/dependency_injection/
builder.rs

1use std::fs;
2use std::sync::Arc;
3use std::time::Duration;
4
5use anyhow::{Context, anyhow};
6use slog::Logger;
7use tokio::sync::{Mutex, RwLock};
8
9use mithril_cardano_node_chain::{
10    chain_observer::{CardanoCliRunner, ChainObserver, ChainObserverBuilder, ChainObserverType},
11    chain_reader::PallasChainReader,
12    chain_scanner::CardanoBlockScanner,
13};
14use mithril_cardano_node_internal_database::{
15    ImmutableFileObserver, ImmutableFileSystemObserver,
16    digesters::CardanoImmutableDigester,
17    digesters::cache::{
18        ImmutableFileDigestCacheProvider, JsonImmutableFileDigestCacheProviderBuilder,
19    },
20    signable_builder::{CardanoDatabaseSignableBuilder, CardanoImmutableFilesFullSignableBuilder},
21};
22use mithril_common::StdResult;
23use mithril_common::api_version::APIVersionProvider;
24use mithril_common::crypto_helper::{
25    KesSigner, KesSignerStandard, OpCert, ProtocolPartyId, SerDeShelleyFileFormat,
26};
27#[cfg(feature = "future_dmq")]
28use mithril_common::messages::RegisterSignatureMessageDmq;
29use mithril_common::signable_builder::{
30    CardanoStakeDistributionSignableBuilder, CardanoTransactionsSignableBuilder,
31    MithrilSignableBuilderService, MithrilStakeDistributionSignableBuilder,
32    SignableBuilderServiceDependencies,
33};
34
35use mithril_era::{EraChecker, EraReader};
36use mithril_signed_entity_lock::SignedEntityTypeLock;
37use mithril_signed_entity_preloader::CardanoTransactionsPreloader;
38use mithril_ticker::{MithrilTickerService, TickerService};
39
40use mithril_persistence::database::repository::CardanoTransactionRepository;
41use mithril_persistence::database::{ApplicationNodeType, SqlMigration};
42use mithril_persistence::sqlite::{ConnectionBuilder, SqliteConnection, SqliteConnectionPool};
43
44use mithril_protocol_config::http_client::http_impl::HttpMithrilNetworkConfigurationProvider;
45
46#[cfg(feature = "future_dmq")]
47use mithril_dmq::{DmqMessageBuilder, DmqPublisherClientPallas};
48
49use crate::dependency_injection::SignerDependencyContainer;
50#[cfg(feature = "future_dmq")]
51use crate::services::SignaturePublisherDmq;
52use crate::services::{
53    AggregatorHTTPClient, CardanoTransactionsImporter,
54    CardanoTransactionsPreloaderActivationSigner, MithrilEpochService, MithrilSingleSigner,
55    SignaturePublishRetryPolicy, SignaturePublisherDelayer, SignaturePublisherNoop,
56    SignaturePublisherRetrier, SignerCertifierService, SignerSignableSeedBuilder,
57    SignerSignedEntityConfigProvider, SignerUpkeepService, TransactionsImporterByChunk,
58    TransactionsImporterWithPruner, TransactionsImporterWithVacuum,
59};
60use crate::store::MKTreeStoreSqlite;
61use crate::{
62    Configuration, HTTP_REQUEST_TIMEOUT_DURATION, MetricsService, SQLITE_FILE,
63    SQLITE_FILE_CARDANO_TRANSACTION,
64};
65use crate::{
66    database::repository::{ProtocolInitializerRepository, SignedBeaconRepository, StakePoolStore},
67    services::SignaturePublisher,
68};
69
70/// The `DependenciesBuilder` is intended to manage Services instance creation.
71///
72/// The goal of this is to put all this code out of the way of business code.
73pub struct DependenciesBuilder<'a> {
74    config: &'a Configuration,
75    chain_observer_builder: fn(&Configuration) -> StdResult<Arc<dyn ChainObserver>>,
76    immutable_file_observer_builder:
77        fn(&Configuration) -> StdResult<Arc<dyn ImmutableFileObserver>>,
78    root_logger: Logger,
79}
80
81impl<'a> DependenciesBuilder<'a> {
82    /// Create a new `DependenciesBuilder`.
83    pub fn new(config: &'a Configuration, root_logger: Logger) -> Self {
84        let chain_observer_builder: fn(&Configuration) -> StdResult<Arc<dyn ChainObserver>> =
85            |config: &Configuration| {
86                let chain_observer_type = ChainObserverType::Pallas;
87                let cardano_cli_path = &config.cardano_cli_path;
88                let cardano_node_socket_path = &config.cardano_node_socket_path;
89                let cardano_network = &config.get_network().with_context(|| {
90                    "Dependencies Builder can not get Cardano network while building the chain observer"
91                })?;
92                let cardano_cli_runner = &CardanoCliRunner::new(
93                    cardano_cli_path.to_owned(),
94                    cardano_node_socket_path.to_owned(),
95                    cardano_network.to_owned(),
96                );
97
98                let chain_observer_builder = ChainObserverBuilder::new(
99                    &chain_observer_type,
100                    cardano_node_socket_path,
101                    cardano_network,
102                    Some(cardano_cli_runner),
103                );
104
105                chain_observer_builder
106                    .build()
107                    .with_context(|| "Dependencies Builder can not build chain observer")
108            };
109
110        let immutable_file_observer_builder: fn(
111            &Configuration,
112        )
113            -> StdResult<Arc<dyn ImmutableFileObserver>> = |config: &Configuration| {
114            Ok(Arc::new(ImmutableFileSystemObserver::new(
115                &config.db_directory,
116            )))
117        };
118
119        Self {
120            config,
121            chain_observer_builder,
122            immutable_file_observer_builder,
123            root_logger,
124        }
125    }
126
127    /// Override immutable file observer builder.
128    pub fn override_immutable_file_observer_builder(
129        &mut self,
130        builder: fn(&Configuration) -> StdResult<Arc<dyn ImmutableFileObserver>>,
131    ) -> &mut Self {
132        self.immutable_file_observer_builder = builder;
133
134        self
135    }
136
137    /// Return a copy of the root logger.
138    pub fn root_logger(&self) -> Logger {
139        self.root_logger.clone()
140    }
141
142    /// Override default chain observer builder.
143    pub fn override_chain_observer_builder(
144        &mut self,
145        builder: fn(&Configuration) -> StdResult<Arc<dyn ChainObserver>>,
146    ) -> &mut Self {
147        self.chain_observer_builder = builder;
148
149        self
150    }
151
152    /// Compute protocol party id
153    fn compute_protocol_party_id(&self) -> StdResult<ProtocolPartyId> {
154        match &self.config.operational_certificate_path {
155            Some(operational_certificate_path) => {
156                let opcert: OpCert = OpCert::from_file(operational_certificate_path)
157                    .with_context(|| "Could not decode operational certificate")?;
158                Ok(opcert
159                    .compute_protocol_party_id()
160                    .with_context(|| "Could not compute party_id from operational certificate")?)
161            }
162            _ => Ok(self
163                .config
164                .party_id
165                .to_owned()
166                .ok_or(anyhow!("A party_id should at least be provided"))?),
167        }
168    }
169
170    async fn build_digester_cache_provider(
171        &self,
172    ) -> StdResult<Option<Arc<dyn ImmutableFileDigestCacheProvider>>> {
173        if self.config.disable_digests_cache {
174            return Ok(None);
175        }
176
177        let cache_provider = JsonImmutableFileDigestCacheProviderBuilder::new(
178            &self.config.data_stores_directory,
179            &format!("immutables_digests_{}.json", self.config.network),
180        )
181        .should_reset_digests_cache(self.config.reset_digests_cache)
182        .with_logger(self.root_logger())
183        .build()
184        .await?;
185
186        Ok(Some(Arc::new(cache_provider)))
187    }
188
189    /// Build a SQLite connection.
190    pub async fn build_sqlite_connection(
191        &self,
192        sqlite_file_name: &str,
193        migrations: Vec<SqlMigration>,
194    ) -> StdResult<SqliteConnection> {
195        let sqlite_db_path = self.config.get_sqlite_file(sqlite_file_name)?;
196        let connection = ConnectionBuilder::open_file(&sqlite_db_path)
197            .with_node_type(ApplicationNodeType::Signer)
198            .with_migrations(migrations)
199            .with_logger(self.root_logger())
200            .build()
201            .with_context(|| "Database connection initialisation error")?;
202
203        Ok(connection)
204    }
205
206    /// Build dependencies for the Production environment.
207    pub async fn build(&self) -> StdResult<SignerDependencyContainer> {
208        if !self.config.data_stores_directory.exists() {
209            fs::create_dir_all(self.config.data_stores_directory.clone()).with_context(|| {
210                format!(
211                    "Could not create data stores directory: `{}`",
212                    self.config.data_stores_directory.display()
213                )
214            })?;
215        }
216
217        let network = self.config.get_network()?;
218        let sqlite_connection = Arc::new(
219            self.build_sqlite_connection(SQLITE_FILE, crate::database::migration::get_migrations())
220                .await?,
221        );
222        let transaction_sqlite_connection = self
223            .build_sqlite_connection(
224                SQLITE_FILE_CARDANO_TRANSACTION,
225                mithril_persistence::database::cardano_transaction_migration::get_migrations(),
226            )
227            .await?;
228        let sqlite_connection_cardano_transaction_pool = Arc::new(
229            SqliteConnectionPool::build_from_connection(transaction_sqlite_connection),
230        );
231
232        let signed_entity_type_lock = Arc::new(SignedEntityTypeLock::default());
233
234        let protocol_initializer_store = Arc::new(ProtocolInitializerRepository::new(
235            sqlite_connection.clone(),
236            self.config.store_retention_limit.map(|limit| limit as u64),
237        ));
238
239        let digester = Arc::new(CardanoImmutableDigester::new(
240            network.to_string(),
241            self.build_digester_cache_provider().await?,
242            self.root_logger(),
243        ));
244        let stake_store = Arc::new(StakePoolStore::new(
245            sqlite_connection.clone(),
246            self.config.store_retention_limit.map(|limit| limit as u64),
247        ));
248        let chain_observer = {
249            let builder = self.chain_observer_builder;
250            builder(self.config)?
251        };
252        let ticker_service = {
253            let builder = self.immutable_file_observer_builder;
254            Arc::new(MithrilTickerService::new(
255                chain_observer.clone(),
256                builder(self.config)?,
257            ))
258        };
259
260        let era_reader = Arc::new(EraReader::new(
261            self.config.build_era_reader_adapter(chain_observer.clone())?,
262        ));
263        let era_epoch_token = era_reader
264            .read_era_epoch_token(ticker_service.get_current_epoch().await?)
265            .await?;
266        let era_checker = Arc::new(EraChecker::new(
267            era_epoch_token.get_current_supported_era()?,
268            era_epoch_token.get_current_epoch(),
269        ));
270
271        let api_version_provider = Arc::new(APIVersionProvider::new(era_checker.clone()));
272        let aggregator_client = Arc::new(AggregatorHTTPClient::new(
273            self.config.aggregator_endpoint.clone(),
274            self.config.relay_endpoint.clone(),
275            api_version_provider.clone(),
276            Some(Duration::from_millis(HTTP_REQUEST_TIMEOUT_DURATION)),
277            self.root_logger(),
278        ));
279
280        let cardano_immutable_snapshot_builder =
281            Arc::new(CardanoImmutableFilesFullSignableBuilder::new(
282                digester.clone(),
283                &self.config.db_directory,
284                self.root_logger(),
285            ));
286        let mithril_stake_distribution_signable_builder =
287            Arc::new(MithrilStakeDistributionSignableBuilder::default());
288        let transaction_store = Arc::new(CardanoTransactionRepository::new(
289            sqlite_connection_cardano_transaction_pool.clone(),
290        ));
291        let chain_block_reader = PallasChainReader::new(
292            &self.config.cardano_node_socket_path,
293            network,
294            self.root_logger(),
295        );
296        let block_scanner = Arc::new(CardanoBlockScanner::new(
297            Arc::new(Mutex::new(chain_block_reader)),
298            self.config
299                .cardano_transactions_block_streamer_max_roll_forwards_per_poll,
300            self.root_logger(),
301        ));
302        let transactions_importer = Arc::new(CardanoTransactionsImporter::new(
303            block_scanner,
304            transaction_store.clone(),
305            self.root_logger(),
306        ));
307        // Wrap the transaction importer with decorator to prune the transactions after import
308        let transactions_importer = Arc::new(TransactionsImporterWithPruner::new(
309            self.config
310                .enable_transaction_pruning
311                .then_some(self.config.network_security_parameter),
312            transaction_store.clone(),
313            transactions_importer,
314            self.root_logger(),
315        ));
316        // Wrap the transaction importer with decorator to chunk its workload, so it prunes
317        // transactions after each chunk, reducing the storage footprint
318        let state_machine_transactions_importer = Arc::new(TransactionsImporterByChunk::new(
319            transaction_store.clone(),
320            transactions_importer.clone(),
321            self.config.transactions_import_block_chunk_size,
322            self.root_logger(),
323        ));
324        // For the preloader, we want to vacuum the database after each chunk, to reclaim disk space
325        // earlier than with just auto_vacuum (that execute only after the end of all import).
326        let preloader_transactions_importer = Arc::new(TransactionsImporterByChunk::new(
327            transaction_store.clone(),
328            Arc::new(TransactionsImporterWithVacuum::new(
329                sqlite_connection_cardano_transaction_pool.clone(),
330                transactions_importer.clone(),
331                self.root_logger(),
332            )),
333            self.config.transactions_import_block_chunk_size,
334            self.root_logger(),
335        ));
336        let block_range_root_retriever = transaction_store.clone();
337        let cardano_transactions_builder = Arc::new(CardanoTransactionsSignableBuilder::<
338            MKTreeStoreSqlite,
339        >::new(
340            state_machine_transactions_importer,
341            block_range_root_retriever,
342        ));
343        let cardano_stake_distribution_signable_builder = Arc::new(
344            CardanoStakeDistributionSignableBuilder::new(stake_store.clone()),
345        );
346        let cardano_database_signable_builder = Arc::new(CardanoDatabaseSignableBuilder::new(
347            digester.clone(),
348            &self.config.db_directory,
349            self.root_logger(),
350        ));
351        let epoch_service = Arc::new(RwLock::new(MithrilEpochService::new(
352            stake_store.clone(),
353            protocol_initializer_store.clone(),
354            self.root_logger(),
355        )));
356        let single_signer = Arc::new(MithrilSingleSigner::new(
357            self.compute_protocol_party_id()?,
358            epoch_service.clone(),
359            self.root_logger(),
360        ));
361        let signable_seed_builder_service = Arc::new(SignerSignableSeedBuilder::new(
362            epoch_service.clone(),
363            protocol_initializer_store.clone(),
364        ));
365        let signable_builders_dependencies = SignableBuilderServiceDependencies::new(
366            mithril_stake_distribution_signable_builder,
367            cardano_immutable_snapshot_builder,
368            cardano_transactions_builder,
369            cardano_stake_distribution_signable_builder,
370            cardano_database_signable_builder,
371        );
372        let signable_builder_service = Arc::new(MithrilSignableBuilderService::new(
373            signable_seed_builder_service,
374            signable_builders_dependencies,
375            self.root_logger(),
376        ));
377        let metrics_service = Arc::new(MetricsService::new(self.root_logger())?);
378        let network_configuration_service = Arc::new(HttpMithrilNetworkConfigurationProvider::new(
379            self.config.aggregator_endpoint.clone(),
380            self.config.relay_endpoint.clone(),
381            api_version_provider.clone(),
382            self.root_logger(),
383        ));
384        let preloader_activation = CardanoTransactionsPreloaderActivationSigner::new(
385            network_configuration_service.clone(),
386        );
387        let cardano_transactions_preloader = Arc::new(CardanoTransactionsPreloader::new(
388            signed_entity_type_lock.clone(),
389            preloader_transactions_importer,
390            self.config.preload_security_parameter,
391            chain_observer.clone(),
392            self.root_logger(),
393            Arc::new(preloader_activation),
394        ));
395        let signed_beacon_repository = Arc::new(SignedBeaconRepository::new(
396            sqlite_connection.clone(),
397            self.config.store_retention_limit.map(|limit| limit as u64),
398        ));
399        let upkeep_service = Arc::new(SignerUpkeepService::new(
400            sqlite_connection.clone(),
401            sqlite_connection_cardano_transaction_pool,
402            signed_entity_type_lock.clone(),
403            vec![
404                signed_beacon_repository.clone(),
405                stake_store.clone(),
406                protocol_initializer_store.clone(),
407            ],
408            self.root_logger(),
409        ));
410
411        let kes_signer = match (
412            &self.config.kes_secret_key_path,
413            &self.config.operational_certificate_path,
414        ) {
415            (Some(kes_secret_key_path), Some(operational_certificate_path)) => {
416                Some(Arc::new(KesSignerStandard::new(
417                    kes_secret_key_path.clone(),
418                    operational_certificate_path.clone(),
419                )) as Arc<dyn KesSigner>)
420            }
421            (Some(_), None) | (None, Some(_)) => {
422                return Err(anyhow!(
423                    "kes_secret_key and operational_certificate are both mandatory".to_string(),
424                ));
425            }
426            _ => None,
427        };
428
429        #[cfg(feature = "future_dmq")]
430        let signature_publisher = {
431            let first_publisher = SignaturePublisherRetrier::new(
432                match &self.config.dmq_node_socket_path {
433                    Some(dmq_node_socket_path) => {
434                        let cardano_network = &self.config.get_network()?;
435                        let dmq_message_builder = DmqMessageBuilder::new(
436                            kes_signer
437                                .clone()
438                                .ok_or(anyhow!("A KES signer is mandatory to sign DMQ messages"))?,
439                            chain_observer.clone(),
440                        );
441                        Arc::new(SignaturePublisherDmq::new(Arc::new(
442                            DmqPublisherClientPallas::<RegisterSignatureMessageDmq>::new(
443                                dmq_node_socket_path.to_owned(),
444                                *cardano_network,
445                                dmq_message_builder,
446                                self.root_logger(),
447                            ),
448                        ))) as Arc<dyn SignaturePublisher>
449                    }
450                    _ => Arc::new(SignaturePublisherNoop) as Arc<dyn SignaturePublisher>,
451                },
452                SignaturePublishRetryPolicy::never(),
453            );
454
455            let second_publisher = SignaturePublisherRetrier::new(
456                aggregator_client.clone(),
457                SignaturePublishRetryPolicy {
458                    attempts: self.config.signature_publisher_config.retry_attempts,
459                    delay_between_attempts: Duration::from_millis(
460                        self.config.signature_publisher_config.retry_delay_ms,
461                    ),
462                },
463            );
464
465            if self.config.signature_publisher_config.skip_delayer {
466                Arc::new(first_publisher) as Arc<dyn SignaturePublisher>
467            } else {
468                Arc::new(SignaturePublisherDelayer::new(
469                    Arc::new(first_publisher),
470                    Arc::new(second_publisher),
471                    Duration::from_millis(self.config.signature_publisher_config.delayer_delay_ms),
472                    self.root_logger(),
473                )) as Arc<dyn SignaturePublisher>
474            }
475        };
476
477        #[cfg(not(feature = "future_dmq"))]
478        let signature_publisher = {
479            let first_publisher = SignaturePublisherRetrier::new(
480                Arc::new(SignaturePublisherNoop) as Arc<dyn SignaturePublisher>,
481                SignaturePublishRetryPolicy::never(),
482            );
483
484            let second_publisher = SignaturePublisherRetrier::new(
485                aggregator_client.clone(),
486                SignaturePublishRetryPolicy {
487                    attempts: self.config.signature_publisher_config.retry_attempts,
488                    delay_between_attempts: Duration::from_millis(
489                        self.config.signature_publisher_config.retry_delay_ms,
490                    ),
491                },
492            );
493
494            if self.config.signature_publisher_config.skip_delayer {
495                Arc::new(second_publisher) as Arc<dyn SignaturePublisher>
496            } else {
497                Arc::new(SignaturePublisherDelayer::new(
498                    Arc::new(first_publisher),
499                    Arc::new(second_publisher),
500                    Duration::from_millis(self.config.signature_publisher_config.delayer_delay_ms),
501                    self.root_logger(),
502                )) as Arc<dyn SignaturePublisher>
503            }
504        };
505
506        let certifier = Arc::new(SignerCertifierService::new(
507            signed_beacon_repository,
508            Arc::new(SignerSignedEntityConfigProvider::new(epoch_service.clone())),
509            signed_entity_type_lock.clone(),
510            single_signer.clone(),
511            signature_publisher,
512            self.root_logger(),
513        ));
514
515        let services = SignerDependencyContainer {
516            ticker_service,
517            certificate_handler: aggregator_client,
518            chain_observer,
519            digester,
520            single_signer,
521            stake_store,
522            protocol_initializer_store,
523            era_checker,
524            era_reader,
525            api_version_provider,
526            signable_builder_service,
527            metrics_service,
528            signed_entity_type_lock,
529            cardano_transactions_preloader,
530            upkeep_service,
531            epoch_service,
532            certifier,
533            kes_signer,
534            network_configuration_service,
535        };
536
537        Ok(services)
538    }
539}
540
541#[cfg(test)]
542mod tests {
543    use std::path::PathBuf;
544
545    use mithril_cardano_node_chain::test::double::FakeChainObserver;
546    use mithril_cardano_node_internal_database::test::double::DumbImmutableFileObserver;
547    use mithril_common::test::double::Dummy;
548    use mithril_common::{entities::TimePoint, test::TempDir};
549
550    use crate::test_tools::TestLogger;
551
552    use super::*;
553
554    fn get_test_dir(test_name: &str) -> PathBuf {
555        TempDir::create("signer_service", test_name)
556    }
557
558    #[tokio::test]
559    async fn test_auto_create_stores_directory() {
560        let stores_dir = get_test_dir("test_auto_create_stores_directory").join("stores");
561        let config = Configuration {
562            data_stores_directory: stores_dir.clone(),
563            ..Configuration::new_sample("party-123456")
564        };
565
566        assert!(!stores_dir.exists());
567        let chain_observer_builder: fn(&Configuration) -> StdResult<Arc<dyn ChainObserver>> =
568            |_config| Ok(Arc::new(FakeChainObserver::new(Some(TimePoint::dummy()))));
569        let immutable_file_observer_builder: fn(
570            &Configuration,
571        )
572            -> StdResult<Arc<dyn ImmutableFileObserver>> =
573            |_config: &Configuration| Ok(Arc::new(DumbImmutableFileObserver::default()));
574
575        let mut dependencies_builder = DependenciesBuilder::new(&config, TestLogger::stdout());
576        dependencies_builder
577            .override_chain_observer_builder(chain_observer_builder)
578            .override_immutable_file_observer_builder(immutable_file_observer_builder)
579            .build()
580            .await
581            .expect("service builder build should not fail");
582        assert!(stores_dir.exists());
583    }
584}
mithril_signer/dependency_injection/builder.rs

mithril_signer/dependency_injection/
builder.rs
