Merge pull request #1798 from input-output-hk/jpraynaud/1797/fix-blocking-ctx-database-insertions

Fix: aggregator and signer blocked during Cardano transactions import

Author: jpraynaud

CHANGELOG.md

@@ -28,6 +28,7 @@ As a minor extension, we have adopted a slightly different versioning convention
   - Chunk the Cardano transactions import in `mithril-signer` to reduce disk footprint by running the pruning process more frequently.
   - Add a database connection pool on the Cardano transaction repository for increased performances of the prover.
   - Import Cardano transactions with Chain Sync mini protocol and Pallas chain reader.
+  - Avoid aggregator and signer being blocked when importing the Cardano transactions.
 
 - Crates versions:
 

mithril-aggregator/src/services/cardano_transactions_importer.rs

@@ -5,6 +5,7 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use slog::{debug, Logger};
+use tokio::{runtime::Handle, task};
 
 use mithril_common::cardano_block_scanner::{BlockScanner, ChainScannedBlocks};
 use mithril_common::crypto_helper::{MKTree, MKTreeNode};
@@ -174,13 +175,25 @@ impl CardanoTransactionsImporter {
             .store_block_range_roots(block_ranges_with_merkle_root)
             .await
     }
+
+    async fn import_transactions_and_block_ranges(
+        &self,
+        up_to_beacon: BlockNumber,
+    ) -> StdResult<()> {
+        self.import_transactions(up_to_beacon).await?;
+        self.import_block_ranges().await
+    }
 }
 
 #[async_trait]
 impl TransactionsImporter for CardanoTransactionsImporter {
     async fn import(&self, up_to_beacon: BlockNumber) -> StdResult<()> {
-        self.import_transactions(up_to_beacon).await?;
-        self.import_block_ranges().await
+        task::block_in_place(move || {
+            Handle::current().block_on(async move {
+                self.import_transactions_and_block_ranges(up_to_beacon)
+                    .await
+            })
+        })
     }
 }
 
@@ -639,7 +652,7 @@ mod tests {
         );
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "multi_thread")]
     async fn importing_twice_starting_with_nothing_in_a_real_db_should_yield_transactions_in_same_order(
     ) {
         let blocks = vec![
@@ -676,7 +689,7 @@ mod tests {
         assert_eq!(cold_imported_transactions, warm_imported_transactions);
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "multi_thread")]
     async fn when_rollbackward_should_remove_transactions() {
         let connection = cardano_tx_db_connection().unwrap();
         let repository = Arc::new(CardanoTransactionRepository::new(Arc::new(
@@ -719,7 +732,7 @@ mod tests {
         assert_eq!(expected_remaining_transactions, stored_transactions);
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "multi_thread")]
     async fn when_rollbackward_should_remove_block_ranges() {
         let connection = cardano_tx_db_connection().unwrap();
         let repository = Arc::new(CardanoTransactionRepository::new(Arc::new(

mithril-aggregator/tests/create_certificate.rs

@@ -10,7 +10,7 @@ use mithril_common::{
 };
 use test_extensions::{utilities::get_test_dir, ExpectedCertificate, RuntimeTester};
 
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn create_certificate() {
     let protocol_parameters = ProtocolParameters {
         k: 5,

mithril-signer/src/cardano_transactions_importer.rs

@@ -5,6 +5,7 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use slog::{debug, Logger};
+use tokio::{runtime::Handle, task};
 
 use mithril_common::cardano_block_scanner::{BlockScanner, ChainScannedBlocks};
 use mithril_common::crypto_helper::{MKTree, MKTreeNode};
@@ -174,13 +175,25 @@ impl CardanoTransactionsImporter {
             .store_block_range_roots(block_ranges_with_merkle_root)
             .await
     }
+
+    async fn import_transactions_and_block_ranges(
+        &self,
+        up_to_beacon: BlockNumber,
+    ) -> StdResult<()> {
+        self.import_transactions(up_to_beacon).await?;
+        self.import_block_ranges().await
+    }
 }
 
 #[async_trait]
 impl TransactionsImporter for CardanoTransactionsImporter {
     async fn import(&self, up_to_beacon: BlockNumber) -> StdResult<()> {
-        self.import_transactions(up_to_beacon).await?;
-        self.import_block_ranges().await
+        task::block_in_place(move || {
+            Handle::current().block_on(async move {
+                self.import_transactions_and_block_ranges(up_to_beacon)
+                    .await
+            })
+        })
     }
 }
 
@@ -639,7 +652,7 @@ mod tests {
         );
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "multi_thread")]
     async fn importing_twice_starting_with_nothing_in_a_real_db_should_yield_transactions_in_same_order(
     ) {
         let blocks = vec![
@@ -676,7 +689,7 @@ mod tests {
         assert_eq!(cold_imported_transactions, warm_imported_transactions);
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "multi_thread")]
     async fn when_rollbackward_should_remove_transactions() {
         let connection = cardano_tx_db_connection().unwrap();
         let repository = Arc::new(CardanoTransactionRepository::new(Arc::new(
@@ -719,7 +732,7 @@ mod tests {
         assert_eq!(expected_remaining_transactions, stored_transactions);
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "multi_thread")]
     async fn when_rollbackward_should_remove_block_ranges() {
         let connection = cardano_tx_db_connection().unwrap();
         let repository = Arc::new(CardanoTransactionRepository::new(Arc::new(

mithril-signer/tests/create_cardano_transaction_single_signature.rs

@@ -9,7 +9,7 @@ use mithril_common::{
 use test_extensions::StateMachineTester;
 
 #[rustfmt::skip]
-#[tokio::test]
+#[tokio::test(flavor = "multi_thread")]
 async fn test_create_cardano_transaction_single_signature() {
     let protocol_parameters = tests_setup::setup_protocol_parameters();
     let fixture = MithrilFixtureBuilder::default()