Merge pull request #1619 from input-output-hk/ensemble/1591/incremental-storage-of-cardano-tx

Incremental storage of Cardano transactions

Author: Alenar

CHANGELOG.md

@@ -13,6 +13,8 @@ As a minor extension, we have adopted a slightly different versioning convention
   - Certificate chain structure has been modified to remove coupling with immutable file number.
   - Client needs to be updated to verify certificate chain.
 
+- Support incremental import for Cardano Transactions instead of scanning the whole immutable database for every signing round.
+
 - Crates versions:
 
 |  Crate  |  Version  |

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-aggregator"
-version = "0.4.53"
+version = "0.4.54"
 description = "A Mithril Aggregator server"
 authors = { workspace = true }
 edition = { workspace = true }

mithril-aggregator/Cargo.toml

@@ -51,5 +51,13 @@ alter table cardano_tx add column block_hash text not null;
 vacuum;
         "#,
         ),
+        // Migration 4
+        // Add index on `block_number` column of `cardano_tx` table
+        SqlMigration::new(
+            4,
+            r#"
+create index block_number_index on cardano_tx(block_number);
+"#,
+        ),
     ]
 }

mithril-aggregator/src/database/cardano_transaction_migration.rs

@@ -3,7 +3,6 @@ use mithril_common::{
         BlockHash, BlockNumber, CardanoDbBeacon, CardanoTransaction, ImmutableFileNumber,
         SlotNumber, TransactionHash,
     },
-    signable_builder::TransactionStore,
     StdResult,
 };
 use mithril_persistence::sqlite::{
@@ -16,7 +15,7 @@ use async_trait::async_trait;
 use sqlite::{Row, Value};
 use std::{iter::repeat, sync::Arc};
 
-use crate::services::TransactionsRetriever;
+use crate::services::{TransactionStore, TransactionsRetriever};
 
 /// Cardano Transaction record is the representation of a cardano transaction.
 #[derive(Debug, PartialEq, Clone)]
@@ -126,11 +125,11 @@ impl<'client> CardanoTransactionProvider<'client> {
 
     pub(crate) fn get_transaction_up_to_beacon_condition(
         &self,
-        beacon: &CardanoDbBeacon,
+        beacon: ImmutableFileNumber,
     ) -> WhereCondition {
         WhereCondition::new(
             "immutable_file_number <= ?*",
-            vec![Value::Integer(beacon.immutable_file_number as i64)],
+            vec![Value::Integer(beacon as i64)],
         )
     }
 }
@@ -236,7 +235,7 @@ impl CardanoTransactionRepository {
     /// chronological order.
     pub async fn get_transactions_up_to(
         &self,
-        beacon: &CardanoDbBeacon,
+        beacon: ImmutableFileNumber,
     ) -> StdResult<Vec<CardanoTransactionRecord>> {
         let provider = CardanoTransactionProvider::new(&self.connection);
         let filters = provider.get_transaction_up_to_beacon_condition(beacon);
@@ -280,12 +279,15 @@ impl CardanoTransactionRepository {
     }
 
     /// Create new [CardanoTransactionRecord]s in the database.
-    pub async fn create_transactions(
+    pub async fn create_transactions<T: Into<CardanoTransactionRecord>>(
         &self,
-        transactions: Vec<CardanoTransactionRecord>,
+        transactions: Vec<T>,
     ) -> StdResult<Vec<CardanoTransactionRecord>> {
+        let records: Vec<CardanoTransactionRecord> =
+            transactions.into_iter().map(|tx| tx.into()).collect();
+
         let provider = InsertCardanoTransactionProvider::new(&self.connection);
-        let filters = provider.get_insert_many_condition(transactions)?;
+        let filters = provider.get_insert_many_condition(records)?;
         let cursor = provider.find(filters)?;
 
         Ok(cursor.collect())
@@ -294,25 +296,62 @@ impl CardanoTransactionRepository {
 
 #[async_trait]
 impl TransactionStore for CardanoTransactionRepository {
-    async fn store_transactions(&self, transactions: &[CardanoTransaction]) -> StdResult<()> {
-        let records: Vec<CardanoTransactionRecord> =
-            transactions.iter().map(|tx| tx.to_owned().into()).collect();
-        self.create_transactions(records)
-            .await
-            .with_context(|| "CardanoTransactionRepository can not store transactions")?;
+    async fn get_highest_beacon(&self) -> StdResult<Option<ImmutableFileNumber>> {
+        let sql = "select max(immutable_file_number) as highest from cardano_tx;";
+        match self
+            .connection
+            .prepare(sql)
+            .with_context(|| {
+                format!(
+                    "Prepare query error: SQL=`{}`",
+                    &sql.replace('\n', " ").trim()
+                )
+            })?
+            .iter()
+            .next()
+        {
+            None => Ok(None),
+            Some(row) => {
+                let highest = row?.read::<Option<i64>, _>(0);
+                highest
+                    .map(u64::try_from)
+                    .transpose()
+                    .with_context(||
+                        format!("Integer field max(immutable_file_number) (value={highest:?}) is incompatible with u64 representation.")
+                    )
+            }
+        }
+    }
+
+    async fn get_up_to(&self, beacon: ImmutableFileNumber) -> StdResult<Vec<CardanoTransaction>> {
+        self.get_transactions_up_to(beacon).await.map(|v| {
+            v.into_iter()
+                .map(|record| record.into())
+                .collect::<Vec<CardanoTransaction>>()
+        })
+    }
 
+    async fn store_transactions(&self, transactions: Vec<CardanoTransaction>) -> StdResult<()> {
+        // Chunk transactions to avoid an error when we exceed sqlite binding limitations
+        for transactions_in_chunk in transactions.chunks(100) {
+            self.create_transactions(transactions_in_chunk.to_vec())
+                .await
+                .with_context(|| "CardanoTransactionRepository can not store transactions")?;
+        }
         Ok(())
     }
 }
 
 #[async_trait]
 impl TransactionsRetriever for CardanoTransactionRepository {
     async fn get_up_to(&self, beacon: &CardanoDbBeacon) -> StdResult<Vec<CardanoTransaction>> {
-        self.get_transactions_up_to(beacon).await.map(|v| {
-            v.into_iter()
-                .map(|record| record.into())
-                .collect::<Vec<CardanoTransaction>>()
-        })
+        self.get_transactions_up_to(beacon.immutable_file_number)
+            .await
+            .map(|v| {
+                v.into_iter()
+                    .map(|record| record.into())
+                    .collect::<Vec<CardanoTransaction>>()
+            })
     }
 }
 
@@ -369,10 +408,7 @@ mod tests {
         let connection = Connection::open_thread_safe(":memory:").unwrap();
         let provider = CardanoTransactionProvider::new(&connection);
         let (expr, params) = provider
-            .get_transaction_up_to_beacon_condition(&CardanoDbBeacon {
-                immutable_file_number: 2309,
-                ..CardanoDbBeacon::default()
-            })
+            .get_transaction_up_to_beacon_condition(2309)
             .expand();
 
         assert_eq!("immutable_file_number <= ?1".to_string(), expr);
@@ -522,7 +558,7 @@ mod tests {
             CardanoTransaction::new("tx-hash-456", 11, 51, "block-hash-456", 100),
         ];
         repository
-            .store_transactions(&cardano_transactions)
+            .create_transactions(cardano_transactions)
             .await
             .unwrap();
 
@@ -558,8 +594,8 @@ mod tests {
         let connection = get_connection().await;
         let repository = CardanoTransactionRepository::new(connection.clone());
 
-        let cardano_transactions: Vec<CardanoTransaction> = (20..=40)
-            .map(|i| CardanoTransaction {
+        let cardano_transactions: Vec<CardanoTransactionRecord> = (20..=40)
+            .map(|i| CardanoTransactionRecord {
                 transaction_hash: format!("tx-hash-{i}"),
                 block_number: i % 10,
                 slot_number: i * 100,
@@ -568,33 +604,18 @@ mod tests {
             })
             .collect();
         repository
-            .store_transactions(&cardano_transactions)
-            .await
-            .unwrap();
-
-        let transaction_result = repository
-            .get_up_to(&CardanoDbBeacon::new("".to_string(), 1, 34))
+            .create_transactions(cardano_transactions.clone())
             .await
             .unwrap();
 
+        let transaction_result = repository.get_transactions_up_to(34).await.unwrap();
         assert_eq!(cardano_transactions[0..=14].to_vec(), transaction_result);
 
-        let transaction_result = repository
-            .get_up_to(&CardanoDbBeacon::new("".to_string(), 1, 300))
-            .await
-            .unwrap();
-
-        assert_eq!(
-            cardano_transactions.into_iter().collect::<Vec<_>>(),
-            transaction_result
-        );
-
-        let transaction_result = repository
-            .get_up_to(&CardanoDbBeacon::new("".to_string(), 1, 19))
-            .await
-            .unwrap();
+        let transaction_result = repository.get_transactions_up_to(300).await.unwrap();
+        assert_eq!(cardano_transactions.clone(), transaction_result);
 
-        assert_eq!(Vec::<CardanoTransaction>::new(), transaction_result);
+        let transaction_result = repository.get_transactions_up_to(19).await.unwrap();
+        assert_eq!(Vec::<CardanoTransactionRecord>::new(), transaction_result);
     }
 
     #[tokio::test]
@@ -607,7 +628,7 @@ mod tests {
             CardanoTransaction::new("tx-hash-456".to_string(), 11, 51, "block-hash-456", 100),
         ];
         repository
-            .store_transactions(&cardano_transactions)
+            .create_transactions(cardano_transactions.clone())
             .await
             .unwrap();
 
@@ -638,7 +659,7 @@ mod tests {
             99,
         )];
         repository
-            .store_transactions(&cardano_transactions)
+            .create_transactions(cardano_transactions)
             .await
             .unwrap();
 
@@ -655,4 +676,31 @@ mod tests {
             transaction_result
         );
     }
+
+    #[tokio::test]
+    async fn repository_get_highest_beacon_without_transactions_in_db() {
+        let connection = get_connection().await;
+        let repository = CardanoTransactionRepository::new(connection.clone());
+
+        let highest_beacon = repository.get_highest_beacon().await.unwrap();
+        assert_eq!(None, highest_beacon);
+    }
+
+    #[tokio::test]
+    async fn repository_get_highest_beacon_with_transactions_in_db() {
+        let connection = get_connection().await;
+        let repository = CardanoTransactionRepository::new(connection.clone());
+
+        let cardano_transactions = vec![
+            CardanoTransaction::new("tx-hash-123".to_string(), 10, 50, "block-hash-123", 50),
+            CardanoTransaction::new("tx-hash-456".to_string(), 11, 51, "block-hash-456", 100),
+        ];
+        repository
+            .create_transactions(cardano_transactions)
+            .await
+            .unwrap();
+
+        let highest_beacon = repository.get_highest_beacon().await.unwrap();
+        assert_eq!(Some(100), highest_beacon);
+    }
 }

mithril-aggregator/src/database/provider/cardano_transaction.rs

@@ -6,7 +6,7 @@ use uuid::Uuid;
 use mithril_common::{entities::Epoch, test_utils::fake_keys, StdResult};
 use mithril_persistence::sqlite::SqliteConnection;
 
-use crate::database::{migration::get_migrations, provider::UpdateSingleSignatureRecordProvider};
+use crate::database::provider::UpdateSingleSignatureRecordProvider;
 
 use super::SingleSignatureRecord;
 
@@ -45,7 +45,15 @@ pub fn disable_foreign_key_support(connection: &SqliteConnection) -> StdResult<(
 }
 
 pub fn apply_all_migrations_to_db(connection: &SqliteConnection) -> StdResult<()> {
-    for migration in get_migrations() {
+    for migration in crate::database::migration::get_migrations() {
+        connection.execute(&migration.alterations)?;
+    }
+
+    Ok(())
+}
+
+pub fn apply_all_transactions_db_migrations(connection: &SqliteConnection) -> StdResult<()> {
+    for migration in crate::database::cardano_transaction_migration::get_migrations() {
         connection.execute(&migration.alterations)?;
     }
 

mithril-aggregator/src/database/provider/test_helper.rs

@@ -33,7 +33,7 @@ use mithril_common::{
     signable_builder::{
         CardanoImmutableFilesFullSignableBuilder, CardanoTransactionsSignableBuilder,
         MithrilSignableBuilderService, MithrilStakeDistributionSignableBuilder,
-        SignableBuilderService, TransactionStore,
+        SignableBuilderService,
     },
     TimePointProvider, TimePointProviderImpl,
 };
@@ -57,10 +57,11 @@ use crate::{
     event_store::{EventMessage, EventStore, TransmitterService},
     http_server::routes::router,
     services::{
-        CertifierService, MessageService, MithrilCertifierService, MithrilEpochService,
-        MithrilMessageService, MithrilProverService, MithrilSignedEntityService,
-        MithrilStakeDistributionService, MithrilTickerService, ProverService, SignedEntityService,
-        StakeDistributionService, TickerService,
+        CardanoTransactionsImporter, CertifierService, MessageService, MithrilCertifierService,
+        MithrilEpochService, MithrilMessageService, MithrilProverService,
+        MithrilSignedEntityService, MithrilStakeDistributionService, MithrilTickerService,
+        ProverService, SignedEntityService, StakeDistributionService, TickerService,
+        TransactionStore,
     },
     tools::{CExplorerSignerRetriever, GcpFileUploader, GenesisToolsDependency, SignersImporter},
     AggregatorConfig, AggregatorRunner, AggregatorRuntime, CertificatePendingStore,
@@ -1079,10 +1080,16 @@ impl DependenciesBuilder {
             &self.configuration.db_directory,
             self.get_logger().await?,
         ));
-        let cardano_transactions_builder = Arc::new(CardanoTransactionsSignableBuilder::new(
+        let transactions_importer = Arc::new(CardanoTransactionsImporter::new(
             self.get_transaction_parser().await?,
             self.get_transaction_store().await?,
             &self.configuration.db_directory,
+            // Rescan the last immutable when importing transactions, it may have been partially imported
+            Some(1),
+            self.get_logger().await?,
+        ));
+        let cardano_transactions_builder = Arc::new(CardanoTransactionsSignableBuilder::new(
+            transactions_importer,
             self.get_logger().await?,
         ));
         let signable_builder_service = Arc::new(MithrilSignableBuilderService::new(