chore(trie): Use trie changesets for engine unwinding

crates/cli/commands/src/stage/unwind.rs

@@ -15,10 +15,7 @@ use reth_db::DatabaseEnv;
 use reth_downloaders::{bodies::noop::NoopBodiesDownloader, headers::noop::NoopHeaderDownloader};
 use reth_evm::ConfigureEvm;
 use reth_exex::ExExManagerHandle;
-use reth_provider::{
-    providers::ProviderNodeTypes, BlockExecutionWriter, BlockNumReader, ChainStateBlockReader,
-    ChainStateBlockWriter, ProviderFactory, StaticFileProviderFactory,
-};
+use reth_provider::{providers::ProviderNodeTypes, BlockNumReader, ProviderFactory};
 use reth_stages::{
     sets::{DefaultStages, OfflineStages},
     stages::ExecutionStage,
@@ -60,54 +57,21 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> Command<C>
 
         let components = components(provider_factory.chain_spec());
 
-        let highest_static_file_block = provider_factory
-            .static_file_provider()
-            .get_highest_static_files()
-            .max_block_num()
-            .filter(|highest_static_file_block| *highest_static_file_block > target);
-
-        // Execute a pipeline unwind if the start of the range overlaps the existing static
-        // files. If that's the case, then copy all available data from MDBX to static files, and
-        // only then, proceed with the unwind.
-        //
-        // We also execute a pipeline unwind if `offline` is specified, because we need to only
-        // unwind the data associated with offline stages.
-        if highest_static_file_block.is_some() || self.offline {
-            if self.offline {
-                info!(target: "reth::cli", "Performing an unwind for offline-only data!");
-            }
-
-            if let Some(highest_static_file_block) = highest_static_file_block {
-                info!(target: "reth::cli", ?target, ?highest_static_file_block, "Executing a pipeline unwind.");
-            } else {
-                info!(target: "reth::cli", ?target, "Executing a pipeline unwind.");
-            }
-            info!(target: "reth::cli", prune_config=?config.prune, "Using prune settings");
-
-            // This will build an offline-only pipeline if the `offline` flag is enabled
-            let mut pipeline =
-                self.build_pipeline(config, provider_factory, components.evm_config().clone())?;
-
-            // Move all applicable data from database to static files.
-            pipeline.move_to_static_files()?;
+        if self.offline {
+            info!(target: "reth::cli", "Performing an unwind for offline-only data!");
+        }
 
-            pipeline.unwind(target, None)?;
-        } else {
-            info!(target: "reth::cli", ?target, "Executing a database unwind.");
-            let provider = provider_factory.provider_rw()?;
+        let highest_static_file_block = provider_factory.provider()?.last_block_number()?;
+        info!(target: "reth::cli", ?target, ?highest_static_file_block, prune_config=?config.prune,  "Executing a pipeline unwind.");
 
-            provider
-                .remove_block_and_execution_above(target)
-                .map_err(|err| eyre::eyre!("Transaction error on unwind: {err}"))?;
+        // This will build an offline-only pipeline if the `offline` flag is enabled
+        let mut pipeline =
+            self.build_pipeline(config, provider_factory, components.evm_config().clone())?;
 
-            // update finalized block if needed
-            let last_saved_finalized_block_number = provider.last_finalized_block_number()?;
-            if last_saved_finalized_block_number.is_none_or(|f| f > target) {
-                provider.save_finalized_block_number(target)?;
-            }
+        // Move all applicable data from database to static files.
+        pipeline.move_to_static_files()?;
 
-            provider.commit()?;
-        }
+        pipeline.unwind(target, None)?;
 
         info!(target: "reth::cli", ?target, "Unwound blocks");
 

crates/ethereum/node/tests/e2e/p2p.rs

@@ -91,7 +91,6 @@ async fn e2e_test_send_transactions() -> eyre::Result<()> {
     Ok(())
 }
 
-#[ignore] // TODO(mediocregopher): re-enable as part of https://github.com/paradigmxyz/reth/issues/18517
 #[tokio::test]
 async fn test_long_reorg() -> eyre::Result<()> {
     reth_tracing::init_test_tracing();

crates/stages/stages/benches/setup/mod.rs

@@ -165,7 +165,7 @@ pub(crate) fn txs_testdata(num_blocks: u64) -> TestStageDB {
         db.insert_changesets(transitions, None).unwrap();
 
         let provider_rw = db.factory.provider_rw().unwrap();
-        provider_rw.write_trie_updates(&updates).unwrap();
+        provider_rw.write_trie_updates(updates).unwrap();
         provider_rw.commit().unwrap();
 
         let (transitions, final_state) = random_changeset_range(

crates/stages/stages/src/stages/merkle.rs

@@ -247,7 +247,7 @@ where
                 })?;
             match progress {
                 StateRootProgress::Progress(state, hashed_entries_walked, updates) => {
-                    provider.write_trie_updates(&updates)?;
+                    provider.write_trie_updates(updates)?;
 
                     let mut checkpoint = MerkleCheckpoint::new(
                         to_block,
@@ -290,7 +290,7 @@ where
                     })
                 }
                 StateRootProgress::Complete(root, hashed_entries_walked, updates) => {
-                    provider.write_trie_updates(&updates)?;
+                    provider.write_trie_updates(updates)?;
 
                     entities_checkpoint.processed += hashed_entries_walked as u64;
 
@@ -317,7 +317,7 @@ where
                         error!(target: "sync::stages::merkle", %e, ?current_block_number, ?to_block, "Incremental state root failed! {INVALID_STATE_ROOT_ERROR_MESSAGE}");
                         StageError::Fatal(Box::new(e))
                     })?;
-                provider.write_trie_updates(&updates)?;
+                provider.write_trie_updates(updates)?;
                 final_root = Some(root);
             }
 
@@ -400,7 +400,7 @@ where
             validate_state_root(block_root, SealedHeader::seal_slow(target), input.unwind_to)?;
 
             // Validation passed, apply unwind changes to the database.
-            provider.write_trie_updates(&updates)?;
+            provider.write_trie_updates(updates)?;
 
             // Update entities checkpoint to reflect the unwind operation
             // Since we're unwinding, we need to recalculate the total entities at the target block

crates/storage/db-common/src/init.rs

@@ -602,7 +602,7 @@ where
 
         match state_root.root_with_progress()? {
             StateRootProgress::Progress(state, _, updates) => {
-                let updated_len = provider.write_trie_updates(&updates)?;
+                let updated_len = provider.write_trie_updates(updates)?;
                 total_flushed_updates += updated_len;
 
                 trace!(target: "reth::cli",
@@ -622,7 +622,7 @@ where
                 }
             }
             StateRootProgress::Complete(root, _, updates) => {
-                let updated_len = provider.write_trie_updates(&updates)?;
+                let updated_len = provider.write_trie_updates(updates)?;
                 total_flushed_updates += updated_len;
 
                 trace!(target: "reth::cli",