fix: batch rewind operations (#4752)

Description
---
Split rewind DbTx into smaller pieces.

How Has This Been Tested?
---
I did rewind on 20000+ (empty) blocks.

base_layer/core/src/chain_storage/blockchain_database.rs

@@ -1647,18 +1647,12 @@ fn check_for_valid_height<T: BlockchainBackend>(db: &T, height: u64) -> Result<(
 
 /// Removes blocks from the db from current tip to specified height.
 /// Returns the blocks removed, ordered from tip to height.
-fn rewind_to_height<T: BlockchainBackend>(
-    db: &mut T,
-    mut height: u64,
-) -> Result<Vec<Arc<ChainBlock>>, ChainStorageError> {
+fn rewind_to_height<T: BlockchainBackend>(db: &mut T, height: u64) -> Result<Vec<Arc<ChainBlock>>, ChainStorageError> {
     let last_header = db.fetch_last_header()?;
 
-    let mut txn = DbTransaction::new();
-
     // Delete headers
     let last_header_height = last_header.height;
     let metadata = db.fetch_chain_metadata()?;
-    let expected_block_hash = *metadata.best_block();
     let last_block_height = metadata.height_of_longest_chain();
     // We use the cmp::max value here because we'll only delete headers here and leave remaining headers to be deleted
     // with the whole block
@@ -1681,20 +1675,20 @@ fn rewind_to_height<T: BlockchainBackend>(
         );
     }
     // We might have more headers than blocks, so we first see if we need to delete the extra headers.
-    (0..steps_back).for_each(|h| {
+    for h in 0..steps_back {
+        let mut txn = DbTransaction::new();
         info!(
             target: LOG_TARGET,
             "Rewinding headers at height {}",
             last_header_height - h
         );
         txn.delete_header(last_header_height - h);
-    });
-
+        db.write(txn)?;
+    }
     // Delete blocks
     let mut steps_back = last_block_height.saturating_sub(height);
     // No blocks to remove, no need to update the best block
     if steps_back == 0 {
-        db.write(txn)?;
         return Ok(vec![]);
     }
 
@@ -1715,22 +1709,45 @@ fn rewind_to_height<T: BlockchainBackend>(
             effective_pruning_horizon
         );
         steps_back = effective_pruning_horizon;
-        height = 0;
     }
-
     for h in 0..steps_back {
+        let mut txn = DbTransaction::new();
         info!(target: LOG_TARGET, "Deleting block {}", last_block_height - h,);
         let block = fetch_block(db, last_block_height - h, false)?;
         let block = Arc::new(block.try_into_chain_block()?);
         txn.delete_block(*block.hash());
         txn.delete_header(last_block_height - h);
         if !prune_past_horizon && !db.contains(&DbKey::OrphanBlock(*block.hash()))? {
-            // Because we know we will remove blocks we can't recover, this will be a destructive rewind, so we can't
-            // recover from this apart from resync from another peer. Failure here should not be common as
-            // this chain has a valid proof of work that has been tested at this point in time.
+            // Because we know we will remove blocks we can't recover, this will be a destructive rewind, so we
+            // can't recover from this apart from resync from another peer. Failure here
+            // should not be common as this chain has a valid proof of work that has been
+            // tested at this point in time.
             txn.insert_chained_orphan(block.clone());
         }
         removed_blocks.push(block);
+        // Set best block to one before, to keep DB consistent. Or if we reached pruned horizon, set best block to 0.
+        let chain_header = db.fetch_chain_header_by_height(if prune_past_horizon && h + 1 == steps_back {
+            0
+        } else {
+            last_block_height - h - 1
+        })?;
+        let metadata = db.fetch_chain_metadata()?;
+        let expected_block_hash = *metadata.best_block();
+        txn.set_best_block(
+            chain_header.height(),
+            chain_header.accumulated_data().hash,
+            chain_header.accumulated_data().total_accumulated_difficulty,
+            expected_block_hash,
+            chain_header.timestamp(),
+        );
+        // Update metadata
+        debug!(
+            target: LOG_TARGET,
+            "Updating best block to height (#{}), total accumulated difficulty: {}",
+            chain_header.height(),
+            chain_header.accumulated_data().total_accumulated_difficulty
+        );
+        db.write(txn)?;
     }
 
     if prune_past_horizon {
@@ -1739,34 +1756,18 @@ fn rewind_to_height<T: BlockchainBackend>(
         // We don't have these complete blocks, so we don't push them to the channel for further processing such as the
         // mempool add reorg'ed tx.
         for h in 0..(last_block_height - steps_back) {
+            let mut txn = DbTransaction::new();
             debug!(
                 target: LOG_TARGET,
                 "Deleting blocks and utxos {}",
                 last_block_height - h - steps_back,
             );
             let header = fetch_header(db, last_block_height - h - steps_back)?;
             txn.delete_block(header.hash());
+            db.write(txn)?;
         }
     }
 
-    let chain_header = db.fetch_chain_header_by_height(height)?;
-    // Update metadata
-    debug!(
-        target: LOG_TARGET,
-        "Updating best block to height (#{}), total accumulated difficulty: {}",
-        chain_header.height(),
-        chain_header.accumulated_data().total_accumulated_difficulty
-    );
-
-    txn.set_best_block(
-        chain_header.height(),
-        chain_header.accumulated_data().hash,
-        chain_header.accumulated_data().total_accumulated_difficulty,
-        expected_block_hash,
-        chain_header.timestamp(),
-    );
-    db.write(txn)?;
-
     Ok(removed_blocks)
 }