feat(base_node): base node keeps track of historical reorgs (#3718)

applications/tari_base_node/src/builder.rs

@@ -235,6 +235,7 @@ async fn build_node_context(
         orphan_storage_capacity: config.orphan_storage_capacity,
         pruning_horizon: config.pruning_horizon,
         pruning_interval: config.pruned_mode_cleanup_interval,
+        track_reorgs: config.blockchain_track_reorgs,
     };
     let blockchain_db = BlockchainDatabase::new(
         backend,

applications/tari_base_node/src/command_handler.rs

@@ -1274,6 +1274,38 @@ impl CommandHandler {
         }
         table.print_stdout();
     }
+
+    pub fn list_reorgs(&self) {
+        if !self.config.blockchain_track_reorgs {
+            println!(
+                "Reorg tracking is turned off. Add `track_reorgs = true` to the [base_node] section of your config to \
+                 turn it on."
+            );
+            return;
+        }
+
+        match self.blockchain_db.inner().fetch_all_reorgs() {
+            Ok(reorgs) => {
+                let mut table = Table::new();
+                table.set_titles(vec!["#", "New Tip", "Prev Tip", "Depth", "Timestamp"]);
+
+                for (i, reorg) in reorgs.iter().enumerate() {
+                    table.add_row(row![
+                        i + 1,
+                        format!("#{} ({})", reorg.new_height, reorg.new_hash.to_hex()),
+                        format!("#{} ({})", reorg.prev_height, reorg.prev_hash.to_hex()),
+                        format!("{} added, {} removed", reorg.num_blocks_added, reorg.num_blocks_removed),
+                        reorg.local_time
+                    ]);
+                }
+
+                table.enable_row_count().print_stdout();
+            },
+            Err(e) => {
+                eprintln!("Error fetching reorgs: {}", e);
+            },
+        }
+    }
 }
 
 async fn fetch_banned_peers(pm: &PeerManager) -> Result<Vec<Peer>, PeerManagerError> {

applications/tari_base_node/src/parser.rs

@@ -79,6 +79,7 @@ pub enum BaseNodeCommand {
     HeaderStats,
     BlockTiming,
     CalcTiming,
+    ListReorgs,
     DiscoverPeer,
     GetBlock,
     SearchUtxo,
@@ -251,6 +252,9 @@ impl Parser {
             BlockTiming | CalcTiming => {
                 self.process_block_timing(args).await;
             },
+            ListReorgs => {
+                self.process_list_reorgs().await;
+            },
             GetBlock => {
                 self.process_get_block(args).await;
             },
@@ -381,6 +385,13 @@ impl Parser {
                 println!("block-timing [start height] [end height]");
                 println!("block-timing [number of blocks from chain tip]");
             },
+            ListReorgs => {
+                println!("List tracked reorgs.");
+                println!(
+                    "This feature must be enabled by setting `track_reorgs = true` in the [base_node] section of your \
+                     config."
+                );
+            },
             GetBlock => {
                 println!("Display a block by height or hash:");
                 println!("get-block [height or hash of the block] [format]");
@@ -731,4 +742,8 @@ impl Parser {
             .and_then(|s| u64::from_str(s).map_err(|_| "new_height must be an integer.")));
         self.command_handler.lock().await.rewind_blockchain(new_height);
     }
+
+    async fn process_list_reorgs(&self) {
+        self.command_handler.lock().await.list_reorgs();
+    }
 }

applications/tari_base_node/src/recovery.rs

@@ -109,6 +109,7 @@ pub async fn run_recovery(node_config: &GlobalConfig) -> Result<(), anyhow::Erro
         orphan_storage_capacity: node_config.orphan_storage_capacity,
         pruning_horizon: node_config.pruning_horizon,
         pruning_interval: node_config.pruned_mode_cleanup_interval,
+        track_reorgs: false,
     };
     let db = BlockchainDatabase::new(
         main_db,

applications/tari_base_node/src/table.rs

@@ -27,6 +27,7 @@ pub struct Table<'t, 's> {
     titles: Option<Vec<&'t str>>,
     rows: Vec<Vec<String>>,
     delim_str: &'s str,
+    is_row_count_enabled: bool,
 }
 
 impl<'t, 's> Table<'t, 's> {
@@ -35,13 +36,19 @@ impl<'t, 's> Table<'t, 's> {
             titles: None,
             rows: Vec::new(),
             delim_str: "|",
+            is_row_count_enabled: false,
         }
     }
 
     pub fn set_titles(&mut self, titles: Vec<&'t str>) {
         self.titles = Some(titles);
     }
 
+    pub fn enable_row_count(&mut self) -> &mut Self {
+        self.is_row_count_enabled = true;
+        self
+    }
+
     pub fn add_row(&mut self, row: Vec<String>) {
         self.rows.push(row);
     }
@@ -55,11 +62,16 @@ impl<'t, 's> Table<'t, 's> {
             self.render_rows(out)?;
             out.write_all(b"\n")?;
         }
+        if self.is_row_count_enabled {
+            out.write_all(format!("\n{} row(s)\n", self.rows.len()).as_bytes())?;
+        }
         Ok(())
     }
 
     pub fn print_stdout(&self) {
-        self.render(&mut io::stdout()).unwrap();
+        let mut stdout = io::stdout();
+        self.render(&mut stdout).unwrap();
+        stdout.flush().unwrap();
     }
 
     fn col_width(&self, idx: usize) -> usize {

base_layer/core/src/chain_storage/blockchain_backend.rs

@@ -27,6 +27,7 @@ use crate::{
         DbValue,
         HorizonData,
         MmrTree,
+        Reorg,
         UtxoMinedInfo,
     },
     transactions::transaction::{TransactionInput, TransactionKernel},
@@ -203,4 +204,7 @@ pub trait BlockchainBackend: Send + Sync {
 
     /// Check if a block hash is in the bad block list
     fn bad_block_exists(&self, block_hash: HashOutput) -> Result<bool, ChainStorageError>;
+
+    /// Fetches all tracked reorgs
+    fn fetch_all_reorgs(&self) -> Result<Vec<Reorg>, ChainStorageError>;
 }

base_layer/core/src/chain_storage/blockchain_database.rs

@@ -73,6 +73,7 @@ use crate::{
         MmrTree,
         Optional,
         OrNotFound,
+        Reorg,
         TargetDifficulties,
     },
     common::rolling_vec::RollingVec,
@@ -96,6 +97,7 @@ pub struct BlockchainDatabaseConfig {
     pub orphan_storage_capacity: usize,
     pub pruning_horizon: u64,
     pub pruning_interval: u64,
+    pub track_reorgs: bool,
 }
 
 impl Default for BlockchainDatabaseConfig {
@@ -104,6 +106,7 @@ impl Default for BlockchainDatabaseConfig {
             orphan_storage_capacity: BLOCKCHAIN_DATABASE_ORPHAN_STORAGE_CAPACITY,
             pruning_horizon: BLOCKCHAIN_DATABASE_PRUNING_HORIZON,
             pruning_interval: BLOCKCHAIN_DATABASE_PRUNED_MODE_PRUNING_INTERVAL,
+            track_reorgs: false,
         }
     }
 }
@@ -246,6 +249,11 @@ where B: BlockchainBackend
             );
             blockchain_db.store_pruning_horizon(config.pruning_horizon)?;
         }
+
+        if !config.track_reorgs {
+            blockchain_db.clear_all_reorgs()?;
+        }
+
         Ok(blockchain_db)
     }
 
@@ -875,6 +883,7 @@ where B: BlockchainBackend
         let mut db = self.db_write_access()?;
         let block_add_result = add_block(
             &mut *db,
+            &self.config,
             &*self.validators.block,
             &*self.validators.header,
             self.consensus_manager.chain_strength_comparer(),
@@ -1110,6 +1119,18 @@ where B: BlockchainBackend
         let lock = self.db_read_access()?;
         lock.fetch_total_size_stats()
     }
+
+    pub fn fetch_all_reorgs(&self) -> Result<Vec<Reorg>, ChainStorageError> {
+        let db = self.db_read_access()?;
+        db.fetch_all_reorgs()
+    }
+
+    pub fn clear_all_reorgs(&self) -> Result<(), ChainStorageError> {
+        let mut db = self.db_write_access()?;
+        let mut txn = DbTransaction::new();
+        txn.clear_all_reorgs();
+        db.write(txn)
+    }
 }
 
 fn unexpected_result<T>(req: DbKey, res: DbValue) -> Result<T, ChainStorageError> {
@@ -1304,6 +1325,7 @@ fn fetch_orphan<T: BlockchainBackend>(db: &T, hash: BlockHash) -> Result<Block,
 
 fn add_block<T: BlockchainBackend>(
     db: &mut T,
+    config: &BlockchainDatabaseConfig,
     block_validator: &dyn PostOrphanBodyValidation<T>,
     header_validator: &dyn HeaderValidation<T>,
     chain_strength_comparer: &dyn ChainStrengthComparer,
@@ -1316,6 +1338,7 @@ fn add_block<T: BlockchainBackend>(
     }
     handle_possible_reorg(
         db,
+        config,
         block_validator,
         header_validator,
         difficulty_calculator,
@@ -1677,6 +1700,7 @@ fn rewind_to_hash<T: BlockchainBackend>(
 // is reorganised if necessary.
 fn handle_possible_reorg<T: BlockchainBackend>(
     db: &mut T,
+    config: &BlockchainDatabaseConfig,
     block_validator: &dyn PostOrphanBodyValidation<T>,
     header_validator: &dyn HeaderValidation<T>,
     difficulty_calculator: &DifficultyCalculator,
@@ -1790,6 +1814,14 @@ fn handle_possible_reorg<T: BlockchainBackend>(
     // reorg is required when any blocks are removed or more than one are added
     // see https://github.com/tari-project/tari/issues/2101
     if num_removed_blocks > 0 || num_added_blocks > 1 {
+        if config.track_reorgs {
+            let mut txn = DbTransaction::new();
+            txn.insert_reorg(Reorg::from_reorged_blocks(&reorg_chain, &removed_blocks));
+            if let Err(e) = db.write(txn) {
+                error!(target: LOG_TARGET, "Failed to track reorg: {}", e);
+            }
+        }
+
         log!(
             target: LOG_TARGET,
             if num_removed_blocks > 1 {
@@ -2713,6 +2745,7 @@ mod test {
         // Add true orphans
         let result = handle_possible_reorg(
             &mut *access,
+            &Default::default(),
             &mock_validator,
             &mock_validator,
             &db.difficulty_calculator,
@@ -2725,6 +2758,7 @@ mod test {
         // Test adding a duplicate orphan
         let result = handle_possible_reorg(
             &mut *access,
+            &Default::default(),
             &mock_validator,
             &mock_validator,
             &db.difficulty_calculator,
@@ -2736,6 +2770,7 @@ mod test {
 
         let result = handle_possible_reorg(
             &mut *access,
+            &Default::default(),
             &mock_validator,
             &mock_validator,
             &db.difficulty_calculator,
@@ -2750,6 +2785,7 @@ mod test {
 
         let result = handle_possible_reorg(
             &mut *access,
+            &Default::default(),
             &mock_validator,
             &mock_validator,
             &db.difficulty_calculator,
@@ -2784,6 +2820,7 @@ mod test {
         // Add true orphans
         let result = handle_possible_reorg(
             &mut *access,
+            &Default::default(),
             &mock_validator,
             &mock_validator,
             &db.difficulty_calculator,
@@ -2795,6 +2832,7 @@ mod test {
 
         let _ = handle_possible_reorg(
             &mut *access,
+            &Default::default(),
             &MockValidator::new(false),
             &mock_validator,
             &db.difficulty_calculator,
@@ -2995,6 +3033,7 @@ mod test {
 
     struct TestHarness {
         db: BlockchainDatabase<TempDatabase>,
+        config: BlockchainDatabaseConfig,
         chain_strength_comparer: Box<dyn ChainStrengthComparer>,
         difficulty_calculator: DifficultyCalculator,
         post_orphan_body_validator: Box<dyn PostOrphanBodyValidation<TempDatabase>>,
@@ -3019,6 +3058,7 @@ mod test {
             let chain_strength_comparer = strongest_chain().by_sha3_difficulty().build();
             Self {
                 db,
+                config: Default::default(),
                 chain_strength_comparer,
                 difficulty_calculator,
                 header_validator,
@@ -3034,6 +3074,7 @@ mod test {
             let mut access = self.db_write_access();
             handle_possible_reorg(
                 &mut *access,
+                &self.config,
                 &*self.post_orphan_body_validator,
                 &*self.header_validator,
                 &self.difficulty_calculator,

base_layer/core/src/chain_storage/db_transaction.rs

@@ -35,7 +35,7 @@ use tari_crypto::tari_utilities::{
 
 use crate::{
     blocks::{Block, BlockHeader, BlockHeaderAccumulatedData, ChainBlock, ChainHeader, UpdateBlockAccumulatedData},
-    chain_storage::{error::ChainStorageError, HorizonData},
+    chain_storage::{error::ChainStorageError, HorizonData, Reorg},
     transactions::transaction::{TransactionKernel, TransactionOutput},
 };
 
@@ -270,6 +270,16 @@ impl DbTransaction {
         self.operations
             .push(WriteOperation::InsertMoneroSeedHeight(monero_seed, height));
     }
+
+    pub fn insert_reorg(&mut self, reorg: Reorg) -> &mut Self {
+        self.operations.push(WriteOperation::InsertReorg { reorg });
+        self
+    }
+
+    pub fn clear_all_reorgs(&mut self) -> &mut Self {
+        self.operations.push(WriteOperation::ClearAllReorgs);
+        self
+    }
 }
 
 #[derive(Debug)]
@@ -338,6 +348,10 @@ pub enum WriteOperation {
     SetHorizonData {
         horizon_data: HorizonData,
     },
+    InsertReorg {
+        reorg: Reorg,
+    },
+    ClearAllReorgs,
 }
 
 impl fmt::Display for WriteOperation {
@@ -426,6 +440,8 @@ impl fmt::Display for WriteOperation {
             DeleteOrphan(hash) => write!(f, "Delete orphan with hash: {}", hash.to_hex()),
             InsertBadBlock { hash, height } => write!(f, "Insert bad block #{} {}", height, hash.to_hex()),
             SetHorizonData { .. } => write!(f, "Set horizon data"),
+            InsertReorg { .. } => write!(f, "Insert reorg"),
+            ClearAllReorgs => write!(f, "Clear all reorgs"),
         }
     }
 }