Merge pull request #1936 from input-output-hk/jpraynaud/1901-fix-data-inconsistency-cardano-transaction

Fix: data inconsistency in Cardano transactions stores

Author: jpraynaud

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-persistence"
-version = "0.2.26"
+version = "0.2.27"
 description = "Common types, interfaces, and utilities to persist data for Mithril nodes."
 authors = { workspace = true }
 edition = { workspace = true }

internal/mithril-persistence/Cargo.toml

@@ -106,6 +106,17 @@ vacuum;
             r#"
 drop index cardano_tx_immutable_file_number_index;
 alter table cardano_tx drop column immutable_file_number;
+ "#,
+        ),
+        // Migration 9
+        // Truncate Cardano transaction related tables to avoid data inconsistency
+        // with previous versions of the nodes.
+        SqlMigration::new(
+            9,
+            r#"
+delete from cardano_tx;
+delete from block_range_root;
+vacuum;
  "#,
         ),
     ]

internal/mithril-persistence/src/database/cardano_transaction_migration.rs

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

mithril-aggregator/Cargo.toml

@@ -731,6 +731,7 @@ impl DependenciesBuilder {
         let chain_block_reader = PallasChainReader::new(
             &self.configuration.cardano_node_socket_path,
             self.configuration.get_network()?,
+            self.get_logger()?,
         );
 
         Ok(Arc::new(Mutex::new(chain_block_reader)))

mithril-aggregator/src/dependency_injection/builder.rs

@@ -7,6 +7,7 @@ use pallas_network::{
     miniprotocols::chainsync::{BlockContent, NextResponse},
 };
 use pallas_traverse::MultiEraBlock;
+use slog::{debug, Logger};
 
 use crate::{cardano_block_scanner::ScannedBlock, entities::ChainPoint, CardanoNetwork, StdResult};
 
@@ -17,15 +18,17 @@ pub struct PallasChainReader {
     socket: PathBuf,
     network: CardanoNetwork,
     client: Option<NodeClient>,
+    logger: Logger,
 }
 
 impl PallasChainReader {
     /// Creates a new `PallasChainReader` with the specified socket and network.
-    pub fn new(socket: &Path, network: CardanoNetwork) -> Self {
+    pub fn new(socket: &Path, network: CardanoNetwork, logger: Logger) -> Self {
         Self {
             socket: socket.to_owned(),
             network,
             client: None,
+            logger,
         }
     }
 
@@ -42,6 +45,7 @@ impl PallasChainReader {
     async fn get_client(&mut self) -> StdResult<&mut NodeClient> {
         if self.client.is_none() {
             self.client = Some(self.new_client().await?);
+            debug!(self.logger, "PallasChainReader connected to a new client");
         }
 
         self.client
@@ -51,13 +55,17 @@ impl PallasChainReader {
 
     /// Intersects the point of the chain with the given point.
     async fn find_intersect_point(&mut self, point: &ChainPoint) -> StdResult<()> {
+        let logger = self.logger.clone();
         let client = self.get_client().await?;
         let chainsync = client.chainsync();
 
         if chainsync.has_agency() {
+            debug!(logger, "PallasChainReader has agency, finding intersect point..."; "point" => ?point);
             chainsync
                 .find_intersect(vec![point.to_owned().into()])
                 .await?;
+        } else {
+            debug!(logger, "PallasChainReader doesn't have agency, no need to find intersect point";);
         }
 
         Ok(())
@@ -130,6 +138,7 @@ mod tests {
 
     use super::*;
 
+    use crate::test_utils::TestLogger;
     use crate::{entities::BlockNumber, test_utils::TempDir};
 
     /// Enum representing the action to be performed by the server.
@@ -253,8 +262,11 @@ mod tests {
         )
         .await;
         let client = tokio::spawn(async move {
-            let mut chain_reader =
-                PallasChainReader::new(socket_path.as_path(), CardanoNetwork::TestNet(10));
+            let mut chain_reader = PallasChainReader::new(
+                socket_path.as_path(),
+                CardanoNetwork::TestNet(10),
+                TestLogger::stdout(),
+            );
 
             chain_reader
                 .set_chain_point(&ChainPoint::from(known_point.clone()))
@@ -285,8 +297,11 @@ mod tests {
         )
         .await;
         let client = tokio::spawn(async move {
-            let mut chain_reader =
-                PallasChainReader::new(socket_path.as_path(), CardanoNetwork::TestNet(10));
+            let mut chain_reader = PallasChainReader::new(
+                socket_path.as_path(),
+                CardanoNetwork::TestNet(10),
+                TestLogger::stdout(),
+            );
 
             chain_reader
                 .set_chain_point(&ChainPoint::from(known_point.clone()))
@@ -317,8 +332,11 @@ mod tests {
         )
         .await;
         let client = tokio::spawn(async move {
-            let mut chain_reader =
-                PallasChainReader::new(socket_path.as_path(), CardanoNetwork::TestNet(10));
+            let mut chain_reader = PallasChainReader::new(
+                socket_path.as_path(),
+                CardanoNetwork::TestNet(10),
+                TestLogger::stdout(),
+            );
 
             chain_reader
                 .set_chain_point(&ChainPoint::from(known_point.clone()))

mithril-common/src/chain_reader/pallas_chain_reader.rs

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-signer"
-version = "0.2.181"
+version = "0.2.182"
 description = "A Mithril Signer"
 authors = { workspace = true }
 edition = { workspace = true }

mithril-signer/Cargo.toml

@@ -264,8 +264,11 @@ impl<'a> DependenciesBuilder<'a> {
         let transaction_store = Arc::new(CardanoTransactionRepository::new(
             sqlite_connection_cardano_transaction_pool.clone(),
         ));
-        let chain_block_reader =
-            PallasChainReader::new(&self.config.cardano_node_socket_path, network);
+        let chain_block_reader = PallasChainReader::new(
+            &self.config.cardano_node_socket_path,
+            network,
+            slog_scope::logger(),
+        );
         let block_scanner = Arc::new(CardanoBlockScanner::new(
             Arc::new(Mutex::new(chain_block_reader)),
             self.config