fix(provider): use storage-aware methods in unwind_trie_state_from (#21561)

crates/e2e-test-utils/tests/rocksdb/main.rs

@@ -469,3 +469,123 @@ async fn test_rocksdb_pending_tx_not_in_storage() -> Result<()> {
 
     Ok(())
 }
+
+/// Reorg with `RocksDB`: verifies that unwind correctly reads changesets from
+/// storage-aware locations (static files vs MDBX) rather than directly from MDBX.
+///
+/// This test exercises `unwind_trie_state_from` which previously failed with
+/// `UnsortedInput` errors because it read changesets directly from MDBX tables
+/// instead of using storage-aware methods that check `storage_changesets_in_static_files`.
+#[tokio::test]
+async fn test_rocksdb_reorg_unwind() -> Result<()> {
+    reth_tracing::init_test_tracing();
+
+    let chain_spec = test_chain_spec();
+    let chain_id = chain_spec.chain().id();
+
+    let (mut nodes, _tasks, _) = E2ETestSetupBuilder::<EthereumNode, _>::new(
+        1,
+        chain_spec.clone(),
+        test_attributes_generator,
+    )
+    .with_tree_config_modifier(|config| config.with_persistence_threshold(0))
+    .build()
+    .await?;
+
+    assert_eq!(nodes.len(), 1);
+
+    // Use two separate wallets to avoid nonce conflicts during reorg
+    let wallets = wallet::Wallet::new(2).with_chain_id(chain_id).wallet_gen();
+    let signer1 = wallets[0].clone();
+    let signer2 = wallets[1].clone();
+    let client = nodes[0].rpc_client().expect("RPC client");
+
+    // Mine block 1 with a transaction from signer1
+    let raw_tx1 =
+        TransactionTestContext::transfer_tx_bytes_with_nonce(chain_id, signer1.clone(), 0).await;
+    let tx_hash1 = nodes[0].rpc.inject_tx(raw_tx1).await?;
+    wait_for_pending_tx(&client, tx_hash1).await;
+
+    let payload1 = nodes[0].advance_block().await?;
+    let block1_hash = payload1.block().hash();
+    assert_eq!(payload1.block().number(), 1);
+
+    // Poll until tx1 appears in RocksDB (ensures persistence happened)
+    let tx_number1 = poll_tx_in_rocksdb(&nodes[0].inner.provider, tx_hash1).await;
+    assert_eq!(tx_number1, 0, "First tx should have tx_number 0");
+
+    // Mine block 2 with transaction from signer1 (nonce 1)
+    let raw_tx2 =
+        TransactionTestContext::transfer_tx_bytes_with_nonce(chain_id, signer1.clone(), 1).await;
+    let tx_hash2 = nodes[0].rpc.inject_tx(raw_tx2).await?;
+    wait_for_pending_tx(&client, tx_hash2).await;
+
+    let payload2 = nodes[0].advance_block().await?;
+    assert_eq!(payload2.block().number(), 2);
+
+    // Poll until tx2 appears in RocksDB
+    let tx_number2 = poll_tx_in_rocksdb(&nodes[0].inner.provider, tx_hash2).await;
+    assert_eq!(tx_number2, 1, "Second tx should have tx_number 1");
+
+    // Mine block 3 with transaction from signer1 (nonce 2)
+    let raw_tx3 =
+        TransactionTestContext::transfer_tx_bytes_with_nonce(chain_id, signer1.clone(), 2).await;
+    let tx_hash3 = nodes[0].rpc.inject_tx(raw_tx3).await?;
+    wait_for_pending_tx(&client, tx_hash3).await;
+
+    let payload3 = nodes[0].advance_block().await?;
+    assert_eq!(payload3.block().number(), 3);
+
+    // Poll until tx3 appears in RocksDB
+    let tx_number3 = poll_tx_in_rocksdb(&nodes[0].inner.provider, tx_hash3).await;
+    assert_eq!(tx_number3, 2, "Third tx should have tx_number 2");
+
+    // Now create an alternate block 2 using signer2 (different wallet, avoids nonce conflict)
+    // Inject a tx from signer2 (nonce 0) before building the alternate block
+    let raw_alt_tx =
+        TransactionTestContext::transfer_tx_bytes_with_nonce(chain_id, signer2.clone(), 0).await;
+    let alt_tx_hash = nodes[0].rpc.inject_tx(raw_alt_tx).await?;
+    wait_for_pending_tx(&client, alt_tx_hash).await;
+
+    // Build an alternate payload (this builds on top of the current head, i.e., block 3)
+    // But we want to reorg back to block 1, so we'll use the payload and then FCU to it
+    let alt_payload = nodes[0].new_payload().await?;
+    let alt_block_hash = nodes[0].submit_payload(alt_payload.clone()).await?;
+
+    // Trigger reorg: make the alternate chain canonical by sending FCU pointing to block 1's hash
+    // as finalized, which should trigger an unwind of blocks 2 and 3
+    // The alt block becomes the new head
+    nodes[0].update_forkchoice(block1_hash, alt_block_hash).await?;
+
+    // Give time for the reorg to complete
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    // Verify we can still query transactions and the chain is consistent
+    // If unwind_trie_state_from failed, this would have errored during reorg
+    let latest: Option<alloy_rpc_types_eth::Block> =
+        client.request("eth_getBlockByNumber", ("latest", false)).await?;
+    let latest = latest.expect("Latest block should exist");
+    // The alt block is at height 4 (on top of block 3)
+    assert!(latest.header.number >= 3, "Should be at height >= 3 after operation");
+
+    // tx1 from block 1 should still be there
+    let tx1: Option<Transaction> = client.request("eth_getTransactionByHash", [tx_hash1]).await?;
+    assert!(tx1.is_some(), "tx1 from block 1 should still be queryable");
+    assert_eq!(tx1.unwrap().block_number, Some(1));
+
+    // Mine another block to verify the chain can continue
+    let raw_tx_final =
+        TransactionTestContext::transfer_tx_bytes_with_nonce(chain_id, signer2.clone(), 1).await;
+    let tx_hash_final = nodes[0].rpc.inject_tx(raw_tx_final).await?;
+    wait_for_pending_tx(&client, tx_hash_final).await;
+
+    let final_payload = nodes[0].advance_block().await?;
+    assert!(final_payload.block().number() > 3, "Should be able to mine block after reorg");
+
+    // Verify tx_final is included
+    let tx_final: Option<Transaction> =
+        client.request("eth_getTransactionByHash", [tx_hash_final]).await?;
+    assert!(tx_final.is_some(), "final tx should be in latest block");
+
+    Ok(())
+}

crates/storage/provider/src/providers/database/provider.rs

@@ -761,11 +761,7 @@ impl<TX: DbTx + DbTxMut + 'static, N: NodeTypesForProvider> DatabaseProvider<TX,
     /// This includes calculating the resulted state root and comparing it with the parent block
     /// state root.
     pub fn unwind_trie_state_from(&self, from: BlockNumber) -> ProviderResult<()> {
-        let changed_accounts = self
-            .tx
-            .cursor_read::<tables::AccountChangeSets>()?
-            .walk_range(from..)?
-            .collect::<Result<Vec<_>, _>>()?;
+        let changed_accounts = self.account_changesets_range(from..)?;
 
         // Unwind account hashes.
         self.unwind_account_hashing(changed_accounts.iter())?;
@@ -773,12 +769,7 @@ impl<TX: DbTx + DbTxMut + 'static, N: NodeTypesForProvider> DatabaseProvider<TX,
         // Unwind account history indices.
         self.unwind_account_history_indices(changed_accounts.iter())?;
 
-        let storage_start = BlockNumberAddress((from, Address::ZERO));
-        let changed_storages = self
-            .tx
-            .cursor_read::<tables::StorageChangeSets>()?
-            .walk_range(storage_start..)?
-            .collect::<Result<Vec<_>, _>>()?;
+        let changed_storages = self.storage_changesets_range(from..)?;
 
         // Unwind storage hashes.
         self.unwind_storage_hashing(changed_storages.iter().copied())?;