Add pruning of cardano transactions

Author: Alenar

internal/mithril-persistence/src/database/provider/cardano_transaction/delete_cardano_transaction.rs

@@ -0,0 +1,139 @@
+use sqlite::Value;
+
+use mithril_common::entities::BlockNumber;
+use mithril_common::StdResult;
+
+use crate::database::record::CardanoTransactionRecord;
+use crate::sqlite::{
+    EntityCursor, Provider, SourceAlias, SqLiteEntity, SqliteConnection, WhereCondition,
+};
+
+/// Query to delete old [CardanoTransactionRecord] from the sqlite database
+pub struct DeleteCardanoTransactionProvider<'conn> {
+    connection: &'conn SqliteConnection,
+}
+
+impl<'conn> Provider<'conn> for DeleteCardanoTransactionProvider<'conn> {
+    type Entity = CardanoTransactionRecord;
+
+    fn get_connection(&'conn self) -> &'conn SqliteConnection {
+        self.connection
+    }
+
+    fn get_definition(&self, condition: &str) -> String {
+        // it is important to alias the fields with the same name as the table
+        // since the table cannot be aliased in a RETURNING statement in SQLite.
+        let projection = Self::Entity::get_projection()
+            .expand(SourceAlias::new(&[("{:cardano_tx:}", "cardano_tx")]));
+
+        format!("delete from cardano_tx where {condition} returning {projection}")
+    }
+}
+
+impl<'conn> DeleteCardanoTransactionProvider<'conn> {
+    /// Create a new instance
+    pub fn new(connection: &'conn SqliteConnection) -> Self {
+        Self { connection }
+    }
+
+    fn get_prune_condition(&self, number_of_block_to_keep: BlockNumber) -> WhereCondition {
+        let number_of_block_to_keep = Value::Integer(number_of_block_to_keep.try_into().unwrap());
+
+        WhereCondition::new(
+            "block_number < ((select max(block_number) from cardano_tx) - ?*)",
+            vec![number_of_block_to_keep],
+        )
+    }
+
+    /// Prune the cardano transaction data after the given number of block.
+    pub fn prune(
+        &self,
+        number_of_block_to_keep: BlockNumber,
+    ) -> StdResult<EntityCursor<CardanoTransactionRecord>> {
+        let filters = self.get_prune_condition(number_of_block_to_keep);
+
+        self.find(filters)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::database::provider::{
+        GetCardanoTransactionProvider, InsertCardanoTransactionProvider,
+    };
+    use crate::database::test_helper::cardano_tx_db_connection;
+    use crate::sqlite::GetAllProvider;
+
+    use super::*;
+
+    fn insert_transactions(connection: &SqliteConnection, records: Vec<CardanoTransactionRecord>) {
+        let provider = InsertCardanoTransactionProvider::new(connection);
+        let condition = provider.get_insert_many_condition(records).unwrap();
+        let mut cursor = provider.find(condition).unwrap();
+        cursor.next().unwrap();
+    }
+
+    fn test_transaction_set() -> Vec<CardanoTransactionRecord> {
+        vec![
+            CardanoTransactionRecord::new("tx-hash-0", 10, 50, "block-hash-10", 1),
+            CardanoTransactionRecord::new("tx-hash-1", 10, 51, "block-hash-10", 1),
+            CardanoTransactionRecord::new("tx-hash-2", 11, 52, "block-hash-11", 1),
+            CardanoTransactionRecord::new("tx-hash-3", 11, 53, "block-hash-11", 1),
+            CardanoTransactionRecord::new("tx-hash-4", 12, 54, "block-hash-12", 1),
+            CardanoTransactionRecord::new("tx-hash-5", 12, 55, "block-hash-12", 1),
+        ]
+    }
+
+    #[test]
+    fn test_prune_work_even_without_transactions_in_db() {
+        let connection = cardano_tx_db_connection().unwrap();
+
+        let prune_provider = DeleteCardanoTransactionProvider::new(&connection);
+        let cursor = prune_provider
+            .prune(100)
+            .expect("pruning shouldn't crash without transactions stored");
+        assert_eq!(0, cursor.count());
+    }
+
+    #[test]
+    fn test_prune_keep_all_data_if_given_block_number_is_larger_than_stored_number_of_block() {
+        let connection = cardano_tx_db_connection().unwrap();
+        insert_transactions(&connection, test_transaction_set());
+
+        let prune_provider = DeleteCardanoTransactionProvider::new(&connection);
+        let cursor = prune_provider.prune(100_000).unwrap();
+        assert_eq!(0, cursor.count());
+
+        let get_provider = GetCardanoTransactionProvider::new(&connection);
+        let cursor = get_provider.get_all().unwrap();
+        assert_eq!(test_transaction_set().len(), cursor.count());
+    }
+
+    #[test]
+    fn test_prune_keep_only_tx_of_last_block_if_given_number_of_block_is_zero() {
+        let connection = cardano_tx_db_connection().unwrap();
+        insert_transactions(&connection, test_transaction_set());
+
+        let prune_provider = DeleteCardanoTransactionProvider::new(&connection);
+        let cursor = prune_provider.prune(0).unwrap();
+        assert_eq!(4, cursor.count());
+
+        let get_provider = GetCardanoTransactionProvider::new(&connection);
+        let cursor = get_provider.get_all().unwrap();
+        assert_eq!(2, cursor.count());
+    }
+
+    #[test]
+    fn test_prune_data_of_older_than_n_blocks() {
+        let connection = cardano_tx_db_connection().unwrap();
+        insert_transactions(&connection, test_transaction_set());
+
+        let prune_provider = DeleteCardanoTransactionProvider::new(&connection);
+        let cursor = prune_provider.prune(1).unwrap();
+        assert_eq!(2, cursor.count());
+
+        let get_provider = GetCardanoTransactionProvider::new(&connection);
+        let cursor = get_provider.get_all().unwrap();
+        assert_eq!(4, cursor.count());
+    }
+}

internal/mithril-persistence/src/database/provider/cardano_transaction/mod.rs

@@ -1,5 +1,7 @@
+mod delete_cardano_transaction;
 mod get_cardano_transaction;
 mod insert_cardano_transaction;
 
+pub use delete_cardano_transaction::*;
 pub use get_cardano_transaction::*;
 pub use insert_cardano_transaction::*;

internal/mithril-persistence/src/database/repository/cardano_transaction_repository.rs

@@ -14,7 +14,7 @@ use mithril_common::signable_builder::BlockRangeRootRetriever;
 use mithril_common::StdResult;
 
 use crate::database::provider::{
-    GetBlockRangeRootProvider, GetCardanoTransactionProvider,
+    DeleteCardanoTransactionProvider, GetBlockRangeRootProvider, GetCardanoTransactionProvider,
     GetIntervalWithoutBlockRangeRootProvider, InsertBlockRangeRootProvider,
     InsertCardanoTransactionProvider,
 };
@@ -273,6 +273,13 @@ impl CardanoTransactionRepository {
 
         Ok(transactions.collect())
     }
+
+    /// Prune the transaction strictly after the given number of block.
+    pub async fn prune_transaction(&self, number_of_block_to_keep: BlockNumber) -> StdResult<()> {
+        let provider = DeleteCardanoTransactionProvider::new(&self.connection);
+        provider.prune(number_of_block_to_keep)?.next();
+        Ok(())
+    }
 }
 
 #[async_trait]
@@ -835,4 +842,36 @@ mod tests {
             );
         }
     }
+
+    #[tokio::test]
+    async fn repository_prune_transactions() {
+        let connection = Arc::new(cardano_tx_db_connection().unwrap());
+        let repository = CardanoTransactionRepository::new(connection);
+
+        // Build transactions with block numbers from 20 to 50
+        let cardano_transactions: Vec<CardanoTransactionRecord> = (20..=50)
+            .map(|i| CardanoTransactionRecord {
+                transaction_hash: format!("tx-hash-{i}"),
+                block_number: i,
+                slot_number: i * 100,
+                block_hash: format!("block-hash-{i}"),
+                immutable_file_number: 1,
+            })
+            .collect();
+
+        repository
+            .create_transactions(cardano_transactions.clone())
+            .await
+            .unwrap();
+
+        let transaction_result = repository.get_all().await.unwrap();
+        assert_eq!(cardano_transactions.len(), transaction_result.len());
+
+        repository.prune_transaction(24).await.unwrap();
+        let transaction_result = repository
+            .get_transactions_in_range_blocks(0..26)
+            .await
+            .unwrap();
+        assert_eq!(Vec::<CardanoTransactionRecord>::new(), transaction_result);
+    }
 }