fix(tree): replace debug_assert with relaxed removal logic (#10634)

Co-authored-by: Roman Krasiuk <[email protected]>

.github/assets/hive/expected_failures_experimental.yaml

@@ -38,7 +38,6 @@ engine-withdrawals:
 # https://github.com/paradigmxyz/reth/issues/8305
 # https://github.com/paradigmxyz/reth/issues/6217
 engine-api:
-  - Re-org to Previously Validated Sidechain Payload (Paris) (reth)
   - Invalid Missing Ancestor ReOrg, StateRoot, EmptyTxs=False, Invalid P9 (Paris) (reth)
   - Invalid Missing Ancestor ReOrg, StateRoot, EmptyTxs=True, Invalid P9 (Paris) (reth)
   - Invalid Missing Ancestor ReOrg, StateRoot, EmptyTxs=False, Invalid P10 (Paris) (reth)
@@ -54,7 +53,6 @@ engine-cancun:
   - Invalid NewPayload, BlobGasUsed, Syncing=True, EmptyTxs=False, DynFeeTxs=False (Cancun) (reth)
   - Invalid NewPayload, Blob Count on BlobGasUsed, Syncing=True, EmptyTxs=False, DynFeeTxs=False (Cancun) (reth)
   - Invalid NewPayload, ExcessBlobGas, Syncing=True, EmptyTxs=False, DynFeeTxs=False (Cancun) (reth)
-  - Re-org to Previously Validated Sidechain Payload (Cancun) (reth)
   - Invalid Missing Ancestor ReOrg, StateRoot, EmptyTxs=False, Invalid P9 (Cancun) (reth)
   - Invalid Missing Ancestor ReOrg, StateRoot, EmptyTxs=True, Invalid P9 (Cancun) (reth)
   - Invalid Missing Ancestor ReOrg, StateRoot, EmptyTxs=False, Invalid P10 (Cancun) (reth)

crates/engine/tree/src/persistence.rs

@@ -88,6 +88,7 @@ where
     }
 
     fn on_remove_blocks_above(&self, new_tip_num: u64) -> Result<(), PersistenceError> {
+        debug!(target: "tree::persistence", ?new_tip_num, "Removing blocks");
         let start_time = Instant::now();
         let provider_rw = self.provider.provider_rw()?;
         let sf_provider = self.provider.static_file_provider();
@@ -100,6 +101,7 @@ where
     }
 
     fn on_save_blocks(&self, blocks: Vec<ExecutedBlock>) -> Result<Option<B256>, PersistenceError> {
+        debug!(target: "tree::persistence", first=?blocks.first().map(|b| b.block.number), last=?blocks.last().map(|b| b.block.number), "Saving range of blocks");
         let start_time = Instant::now();
         let last_block_hash = blocks.last().map(|block| block.block().hash());
 

crates/engine/tree/src/tree/mod.rs

@@ -221,13 +221,25 @@ impl TreeState {
     ///
     /// Canonical blocks below the upper bound will still be removed.
     ///
-    /// NOTE: This assumes that the `finalized_num` is below or equal to the `upper_bound`
+    /// NOTE: if the finalized block is greater than the upper bound, the only blocks that will be
+    /// removed are canonical blocks and sidechains that fork below the `upper_bound`. This is the
+    /// same behavior as if the `finalized_num` were `Some(upper_bound)`.
     pub(crate) fn remove_until(
         &mut self,
         upper_bound: BlockNumber,
         finalized_num: Option<BlockNumber>,
     ) {
-        debug_assert!(Some(upper_bound) >= finalized_num);
+        debug!(target: "engine", ?upper_bound, ?finalized_num, "Removing blocks from the tree");
+
+        // If the finalized num is ahead of the upper bound, and exists, we need to instead ensure
+        // that the only blocks removed, are canonical blocks less than the upper bound
+        // finalized_num.take_if(|finalized| *finalized > upper_bound);
+        let finalized_num = finalized_num.map(|finalized| {
+            let new_finalized_num = finalized.min(upper_bound);
+            debug!(target: "engine", ?new_finalized_num, "Adjusted upper bound");
+            new_finalized_num
+        });
+
         // We want to do two things:
         // * remove canonical blocks that are persisted
         // * remove forks whose root are below the finalized block
@@ -1247,6 +1259,7 @@ where
         let target_number =
             canonical_head_number.saturating_sub(self.config.memory_block_buffer_target());
 
+        debug!(target: "engine", ?last_persisted_number, ?canonical_head_number, ?target_number, ?current_hash, "Returning canonical blocks to persist");
         while let Some(block) = self.state.tree_state.blocks_by_hash.get(&current_hash) {
             if block.block.number <= last_persisted_number {
                 break;