Transfer immutable offsetting responsability to block scanner

Alenar · sfauvel · dlachaume · Alenar · commit dcceb41ebbbb · 2024-06-04T12:04:05.000+02:00
Since the importer doesn't know anymore about immutable file numbers and
the current block scanner role is precisely to fetch them for its
`ImmutableBlockStreamer`.

Co-authored-by: Sébastien Fauvel &lt;sfauvel@users.noreply.github.com&gt;
Co-authored-by: Damien Lachaume &lt;dlachaume@users.noreply.github.com&gt;
diff --git a/mithril-aggregator/src/dependency_injection/builder.rs b/mithril-aggregator/src/dependency_injection/builder.rs
@@ -700,6 +700,8 @@ impl DependenciesBuilder {
                 .get_network()?
                 .compute_allow_unparsable_block(self.configuration.allow_unparsable_block)?,
             self.get_transaction_repository().await?,
+            // Rescan the last immutable when importing transactions, it may have been partially imported
+            Some(1),
         );
 
         Ok(Arc::new(block_scanner))
@@ -1039,8 +1041,6 @@ impl DependenciesBuilder {
             self.get_block_scanner().await?,
             self.get_transaction_repository().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 block_range_root_retriever = self.get_transaction_repository().await?;
diff --git a/mithril-aggregator/src/services/cardano_transactions_importer.rs b/mithril-aggregator/src/services/cardano_transactions_importer.rs
@@ -46,7 +46,6 @@ pub struct CardanoTransactionsImporter {
     block_scanner: Arc<dyn BlockScanner>,
     transaction_store: Arc<dyn TransactionStore>,
     logger: Logger,
-    rescan_offset: Option<usize>,
     dirpath: PathBuf,
 }
 
@@ -60,14 +59,12 @@ impl CardanoTransactionsImporter {
         block_scanner: Arc<dyn BlockScanner>,
         transaction_store: Arc<dyn TransactionStore>,
         dirpath: &Path,
-        rescan_offset: Option<usize>,
         logger: Logger,
     ) -> Self {
         Self {
             block_scanner,
             transaction_store,
             logger,
-            rescan_offset,
             dirpath: dirpath.to_owned(),
         }
     }
@@ -213,7 +210,6 @@ mod tests {
                 scanner,
                 transaction_store,
                 Path::new(""),
-                None,
                 crate::test_tools::logger_for_tests(),
             )
         }
diff --git a/mithril-common/src/cardano_block_scanner/block_scanner.rs b/mithril-common/src/cardano_block_scanner/block_scanner.rs
@@ -32,6 +32,7 @@ pub struct CardanoBlockScanner {
     /// This situation should only happen on the test networks and not on the mainnet.
     allow_unparsable_block: bool,
     lower_bound_finder: Arc<dyn ImmutableLowerBoundFinder>,
+    rescan_offset: Option<usize>,
 }
 
 impl CardanoBlockScanner {
@@ -40,6 +41,7 @@ impl CardanoBlockScanner {
         logger: Logger,
         allow_unparsable_block: bool,
         lower_bound_finder: Arc<dyn ImmutableLowerBoundFinder>,
+        rescan_offset: Option<usize>,
     ) -> Self {
         if allow_unparsable_block {
             warn!(
@@ -51,8 +53,16 @@ impl CardanoBlockScanner {
             logger,
             allow_unparsable_block,
             lower_bound_finder,
+            rescan_offset,
         }
     }
+
+    async fn get_lower_bound(&self) -> StdResult<Option<ImmutableFileNumber>> {
+        let highest = self.lower_bound_finder.find_lower_bound().await?;
+        let rescan_offset = self.rescan_offset.unwrap_or(0);
+        let highest = highest.map(|h| (h + 1).saturating_sub(rescan_offset as u64));
+        Ok(highest)
+    }
 }
 
 #[async_trait]
@@ -63,7 +73,7 @@ impl BlockScanner for CardanoBlockScanner {
         _from: Option<ChainPoint>,
         _until: BlockNumber,
     ) -> StdResult<Box<dyn BlockStreamer>> {
-        let lower_bound = self.lower_bound_finder.find_lower_bound().await?;
+        let lower_bound = self.get_lower_bound().await?;
         let is_in_bounds = |number: ImmutableFileNumber| match lower_bound {
             Some(from) => from <= number,
             None => true,
@@ -114,7 +124,7 @@ mod tests {
             mock.expect_find_lower_bound().returning(|| Ok(None));
         });
         let cardano_transaction_parser =
-            CardanoBlockScanner::new(TestLogger::stdout(), false, lower_bound_finder);
+            CardanoBlockScanner::new(TestLogger::stdout(), false, lower_bound_finder, None);
 
         for until_block_number in [1, 10000] {
             let mut streamer = cardano_transaction_parser
@@ -145,7 +155,7 @@ mod tests {
             mock.expect_find_lower_bound().returning(|| Ok(Some(0)));
         });
         let cardano_transaction_parser =
-            CardanoBlockScanner::new(TestLogger::stdout(), false, lower_bound_finder);
+            CardanoBlockScanner::new(TestLogger::stdout(), false, lower_bound_finder, None);
 
         for until_block_number in [1, 10000] {
             let mut streamer = cardano_transaction_parser
@@ -172,16 +182,20 @@ mod tests {
         let db_path = Path::new("../mithril-test-lab/test_data/immutable/");
         assert!(get_number_of_immutable_chunk_in_dir(db_path) >= 3);
 
-        let test_cases = [(0, None), (1, Some(1))];
-        for (expected, lowest_found_immutable) in test_cases {
+        let test_cases = [
+            (None, 0),
+            // When a lowest immutable file number is found we start from the next immutable (i + 1)
+            (Some(1), 2),
+        ];
+        for (lowest_found_immutable, expected) in test_cases {
             let lower_bound_finder = lower_bound_finder(|mock| {
                 mock.expect_find_lower_bound()
                     .return_once(move || Ok(lowest_found_immutable));
             });
 
             let from = ChainPoint::dummy();
             let cardano_transaction_parser =
-                CardanoBlockScanner::new(TestLogger::stdout(), false, lower_bound_finder);
+                CardanoBlockScanner::new(TestLogger::stdout(), false, lower_bound_finder, None);
 
             let mut streamer = cardano_transaction_parser
                 .scan(db_path, Some(from), 10000000)
@@ -211,6 +225,7 @@ mod tests {
                 TestLogger::file(&log_path),
                 true,
                 lower_bound_finder(|_| {}),
+                None,
             );
         }
 
@@ -232,10 +247,42 @@ mod tests {
                 TestLogger::file(&log_path),
                 false,
                 lower_bound_finder(|_| {}),
+                None,
             );
         }
 
         let log_file = std::fs::read_to_string(&log_path).unwrap();
         assert!(!log_file.contains("The 'allow_unparsable_block' option is activated. This option should only be used on test networks."));
     }
+
+    #[tokio::test]
+    async fn change_parsed_lower_bound_when_rescan_limit_is_set() {
+        fn scanner_with_offset(
+            highest_stored_immutable: ImmutableFileNumber,
+            rescan_offset: ImmutableFileNumber,
+        ) -> CardanoBlockScanner {
+            let mut store = MockImmutableLowerBoundFinder::new();
+            store
+                .expect_find_lower_bound()
+                .returning(move || Ok(Some(highest_stored_immutable)));
+
+            CardanoBlockScanner::new(
+                TestLogger::stdout(),
+                false,
+                Arc::new(store),
+                Some(rescan_offset as usize),
+            )
+        }
+        let scanner = scanner_with_offset(8, 3);
+
+        let from = scanner.get_lower_bound().await.unwrap();
+        // Expected should be: highest_stored_beacon + 1 - rescan_offset
+        assert_eq!(Some(6), from);
+
+        let scanner = scanner_with_offset(5, 10);
+
+        let from = scanner.get_lower_bound().await.unwrap();
+        // If sub overflow it should be 0
+        assert_eq!(Some(0), from);
+    }
 }
diff --git a/mithril-signer/src/cardano_transactions_importer.rs b/mithril-signer/src/cardano_transactions_importer.rs
@@ -46,7 +46,6 @@ pub struct CardanoTransactionsImporter {
     block_scanner: Arc<dyn BlockScanner>,
     transaction_store: Arc<dyn TransactionStore>,
     logger: Logger,
-    rescan_offset: Option<usize>,
     dirpath: PathBuf,
 }
 
@@ -60,14 +59,12 @@ impl CardanoTransactionsImporter {
         block_scanner: Arc<dyn BlockScanner>,
         transaction_store: Arc<dyn TransactionStore>,
         dirpath: &Path,
-        rescan_offset: Option<usize>,
         logger: Logger,
     ) -> Self {
         Self {
             block_scanner,
             transaction_store,
             logger,
-            rescan_offset,
             dirpath: dirpath.to_owned(),
         }
     }
@@ -213,7 +210,6 @@ mod tests {
                 scanner,
                 transaction_store,
                 Path::new(""),
-                None,
                 crate::test_tools::logger_for_tests(),
             )
         }
diff --git a/mithril-signer/src/runtime/runner.rs b/mithril-signer/src/runtime/runner.rs
@@ -547,7 +547,6 @@ mod tests {
             transaction_parser.clone(),
             transaction_store.clone(),
             Path::new(""),
-            None,
             slog_scope::logger(),
         ));
         let block_range_root_retriever = Arc::new(MockBlockRangeRootRetrieverImpl::new());
diff --git a/mithril-signer/src/runtime/signer_services.rs b/mithril-signer/src/runtime/signer_services.rs
@@ -265,13 +265,13 @@ impl<'a> ServiceBuilder for ProductionServiceBuilder<'a> {
                 .get_network()?
                 .compute_allow_unparsable_block(self.config.allow_unparsable_block)?,
             transaction_store.clone(),
+            // Rescan the last immutable when importing transactions, it may have been partially imported
+            Some(1),
         ));
         let transactions_importer = Arc::new(CardanoTransactionsImporter::new(
             block_scanner,
             transaction_store.clone(),
             &self.config.db_directory,
-            // Rescan the last immutable when importing transactions, it may have been partially imported
-            Some(1),
             slog_scope::logger(),
         ));
         // Wrap the transaction importer with decorator to prune the transactions after import
diff --git a/mithril-signer/tests/test_extensions/state_machine_tester.rs b/mithril-signer/tests/test_extensions/state_machine_tester.rs
@@ -164,7 +164,6 @@ impl StateMachineTester {
             transaction_parser.clone(),
             transaction_store.clone(),
             Path::new(""),
-            None,
             slog_scope::logger(),
         ));
         let block_range_root_retriever = transaction_store.clone();

Original file line number	Diff line number	Diff line change
`@@ -46,7 +46,6 @@ pub struct CardanoTransactionsImporter {`
`46`	`46`	`block_scanner: Arc<dyn BlockScanner>,`
`47`	`47`	`transaction_store: Arc<dyn TransactionStore>,`
`48`	`48`	`logger: Logger,`
`49`		`- rescan_offset: Option<usize>,`
`50`	`49`	`dirpath: PathBuf,`
`51`	`50`	`}`
`52`	`51`
`@@ -60,14 +59,12 @@ impl CardanoTransactionsImporter {`
`60`	`59`	`block_scanner: Arc<dyn BlockScanner>,`
`61`	`60`	`transaction_store: Arc<dyn TransactionStore>,`
`62`	`61`	`dirpath: &Path,`
`63`		`- rescan_offset: Option<usize>,`
`64`	`62`	`logger: Logger,`
`65`	`63`	`) -> Self {`
`66`	`64`	`Self {`
`67`	`65`	`block_scanner,`
`68`	`66`	`transaction_store,`
`69`	`67`	`logger,`
`70`		`- rescan_offset,`
`71`	`68`	`dirpath: dirpath.to_owned(),`
`72`	`69`	`}`
`73`	`70`	`}`
`@@ -213,7 +210,6 @@ mod tests {`
`213`	`210`	`scanner,`
`214`	`211`	`transaction_store,`
`215`	`212`	`Path::new(""),`
`216`		`- None,`
`217`	`213`	`crate::test_tools::logger_for_tests(),`
`218`	`214`	`)`
`219`	`215`	`}`