fix(anvil): make fork resets atomic

.changelog/anvil-atomic-fork-reset.md

@@ -0,0 +1,6 @@
+---
+anvil: patch
+---
+
+Made fork resets atomic so failed or concurrent resets cannot expose partially updated chain state,
+and reject resets that would change the node's fixed execution-network family.

crates/anvil/src/eth/api.rs

@@ -762,7 +762,7 @@ impl<N: Network> EthApi<N> {
             let _mining = self.backend.lock_mining().await;
             self.backend.commit_fork_reset(staged).await?;
             self.reset_instance_id();
-            self.pool.clear();
+            self.pool.reset();
             self.fee_history_cache.lock().clear();
         } else {
             let _lifecycle = self.lifecycle_lock.write().await;
@@ -772,7 +772,7 @@ impl<N: Network> EthApi<N> {
             let staged = self.backend.prepare_memory_reset().await?;
             self.backend.commit_memory_reset(staged).await?;
             self.reset_instance_id();
-            self.pool.clear();
+            self.pool.reset();
             self.fee_history_cache.lock().clear();
         }
         Ok(())
@@ -4561,11 +4561,13 @@ impl EthApi<FoundryNetwork> {
 
     /// Mines exactly one block
     pub async fn mine_one(&self) -> Result<()> {
-        let transactions = self.pool.ready_transactions().collect::<Vec<_>>();
-        let outcome = self.backend.mine_block(transactions).await?;
+        let batch = self.pool.mining_batch(None);
+        let Some((generation, outcome)) = self.backend.mine_pool_batch(batch).await? else {
+            return Ok(());
+        };
 
         trace!(target: "node", blocknumber = ?outcome.block_number, "mined block");
-        self.pool.on_mined_block(outcome);
+        self.pool.on_mined_block_at_generation(generation, outcome);
         Ok(())
     }
 
@@ -5048,6 +5050,18 @@ mod tests {
     use super::*;
     use crate::{NodeConfig, spawn};
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn reset_discards_previously_selected_mining_batch() {
+        let (api, _) = spawn(NodeConfig::test()).await;
+        let batch = api.pool.mining_batch(None);
+
+        api.anvil_reset(None).await.unwrap();
+        let best_number = api.backend.best_number();
+
+        assert!(api.backend.mine_pool_batch(batch).await.unwrap().is_none());
+        assert_eq!(api.backend.best_number(), best_number);
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn memory_reset_stages_live_fees_after_active_mining() {
         let (api, _handle) = spawn(NodeConfig::test()).await;

crates/anvil/src/eth/backend/mem/mod.rs

@@ -25,7 +25,9 @@ use crate::{
                 state::{state_root, storage_root, trie_accounts},
                 storage::MinedTransactionReceipt,
             },
-            notifications::{ChainNotification, ChainNotifications, NewBlockNotification},
+            notifications::{
+                ChainNotification, ChainNotifications, FeeHistoryNotification, NewBlockNotification,
+            },
             replay::{
                 ExecutedHistoricalReplay, HistoricalReplayTransaction,
                 PreparedForkTransactionReplay, execute_historical_replay,
@@ -38,7 +40,7 @@ use crate::{
         error::{BlockchainError, ErrDetail, InvalidTransactionError},
         fees::{FeeDetails, FeeManager, MIN_SUGGESTED_PRIORITY_FEE},
         macros::node_info,
-        pool::transactions::PoolTransaction,
+        pool::{MiningBatch, transactions::PoolTransaction},
         preserve_simulation_request_fields,
     },
     mem::{
@@ -208,7 +210,7 @@ use std::{
     path::{Path, PathBuf},
     sync::{
         Arc,
-        atomic::{AtomicBool, Ordering},
+        atomic::{AtomicBool, AtomicU64, Ordering},
     },
     time::Duration,
 };
@@ -1029,6 +1031,8 @@ pub struct Backend<N: Network> {
     /// Listeners for new blocks that get notified when a new block was imported or when logs were
     /// removed from the canonical chain due to a reorg.
     new_block_listeners: Arc<Mutex<Vec<UnboundedSender<ChainNotification>>>>,
+    /// Internal block notifications with reset lifecycle metadata.
+    fee_history_listeners: Arc<Mutex<Vec<UnboundedSender<FeeHistoryNotification>>>>,
     /// Keeps track of active state snapshots at a specific block.
     active_state_snapshots: Arc<Mutex<HashMap<U256, (u64, B256)>>>,
     enable_steps_tracing: bool,
@@ -1047,6 +1051,8 @@ pub struct Backend<N: Network> {
     precompile_factory: Option<Arc<dyn PrecompileFactory>>,
     /// Prevent race conditions during mining
     mining: Arc<tokio::sync::Mutex<()>>,
+    /// Generation used to reject work selected before a reset.
+    reset_generation: Arc<AtomicU64>,
     /// Disable pool balance checks
     disable_pool_balance_checks: bool,
     /// Keeps startup fork-cache rollback armed until startup initialization completes.
@@ -1072,6 +1078,7 @@ impl<N: Network> Clone for Backend<N> {
             fees: self.fees.clone(),
             genesis: self.genesis.clone(),
             new_block_listeners: self.new_block_listeners.clone(),
+            fee_history_listeners: self.fee_history_listeners.clone(),
             active_state_snapshots: self.active_state_snapshots.clone(),
             enable_steps_tracing: self.enable_steps_tracing,
             print_logs: self.print_logs,
@@ -1083,6 +1090,7 @@ impl<N: Network> Clone for Backend<N> {
             slots_in_an_epoch: self.slots_in_an_epoch,
             precompile_factory: self.precompile_factory.clone(),
             mining: self.mining.clone(),
+            reset_generation: self.reset_generation.clone(),
             disable_pool_balance_checks: self.disable_pool_balance_checks,
             startup_fork_cache_user: self.startup_fork_cache_user.clone(),
         }
@@ -2108,6 +2116,22 @@ impl<N: Network> Backend<N> {
         rx
     }
 
+    pub(crate) fn fee_history_notifications(
+        &self,
+    ) -> futures::channel::mpsc::UnboundedReceiver<FeeHistoryNotification> {
+        let (tx, rx) = unbounded();
+        self.fee_history_listeners.lock().push(tx);
+        rx
+    }
+
+    pub(crate) fn reset_generation(&self) -> Arc<AtomicU64> {
+        Arc::clone(&self.reset_generation)
+    }
+
+    pub(crate) fn current_reset_generation(&self) -> u64 {
+        self.reset_generation.load(Ordering::Acquire)
+    }
+
     /// Returns the number of new-block listeners. Closed listeners are pruned lazily on the next
     /// new block notification.
     pub fn new_block_listeners_count(&self) -> usize {
@@ -2120,8 +2144,18 @@ impl<N: Network> Backend<N> {
         // sender half for the set
         self.new_block_listeners.lock().retain(|tx| !tx.is_closed());
 
-        let notification =
-            ChainNotification::Block(NewBlockNotification { hash, header: Arc::new(header) });
+        let header = Arc::new(header);
+        let fee_history_notification = FeeHistoryNotification {
+            hash,
+            header: Arc::clone(&header),
+            generation: self.current_reset_generation(),
+            blob_params: self.fees.blob_params(),
+        };
+        self.fee_history_listeners
+            .lock()
+            .retain(|tx| tx.unbounded_send(fee_history_notification.clone()).is_ok());
+
+        let notification = ChainNotification::Block(NewBlockNotification { hash, header });
 
         self.new_block_listeners
             .lock()
@@ -2183,9 +2217,10 @@ impl<N: Network> Backend<N> {
         let _mining_guard = self.mining.lock().await;
         let next_number = highest.checked_add(1)?;
         if let Some(block) = self.get_block(next_number) {
+            let blob_params = self.simulation_blob_params_at_timestamp(block.header.timestamp);
             Some((
                 block.header.base_fee_per_gas.unwrap_or_default() as u128,
-                block.header.blob_fee(self.blob_params()).unwrap_or_default(),
+                block.header.blob_fee(blob_params).unwrap_or_default(),
             ))
         } else if highest == self.best_number() {
             Some((self.fees().base_fee() as u128, self.fees().base_fee_per_blob_gas()))
@@ -4364,6 +4399,7 @@ impl<N: Network> Backend<N> {
             time: TimeManager::new(start_timestamp),
             cheats: Default::default(),
             new_block_listeners: Default::default(),
+            fee_history_listeners: Default::default(),
             fees,
             genesis,
             active_state_snapshots: Arc::new(Mutex::new(Default::default())),
@@ -4377,6 +4413,7 @@ impl<N: Network> Backend<N> {
             slots_in_an_epoch,
             precompile_factory,
             mining: Arc::new(tokio::sync::Mutex::new(())),
+            reset_generation: Arc::new(AtomicU64::new(0)),
             disable_pool_balance_checks,
             startup_fork_cache_user,
         };
@@ -5384,6 +5421,19 @@ where
         self.do_mine_block(pool_transactions).await
     }
 
+    /// Mines a pool batch unless it was selected before the latest reset.
+    pub(crate) async fn mine_pool_batch(
+        &self,
+        batch: MiningBatch<FoundryTxEnvelope>,
+    ) -> Result<Option<(u64, MinedBlockOutcome<FoundryTxEnvelope>)>, BlockchainError> {
+        let _mining_guard = self.mining.lock().await;
+        if batch.generation != self.current_reset_generation() {
+            return Ok(None);
+        }
+        let outcome = self.do_mine_block_locked(batch.transactions).await?;
+        Ok(Some((batch.generation, outcome)))
+    }
+
     /// Replays a transaction-hash fork prefix before the live pool and miner are created.
     pub(crate) async fn apply_fork_transaction_replay(
         &self,
@@ -5782,6 +5832,13 @@ where
         pool_transactions: Vec<Arc<PoolTransaction<FoundryTxEnvelope>>>,
     ) -> Result<MinedBlockOutcome<FoundryTxEnvelope>, BlockchainError> {
         let _mining_guard = self.mining.lock().await;
+        self.do_mine_block_locked(pool_transactions).await
+    }
+
+    async fn do_mine_block_locked(
+        &self,
+        pool_transactions: Vec<Arc<PoolTransaction<FoundryTxEnvelope>>>,
+    ) -> Result<MinedBlockOutcome<FoundryTxEnvelope>, BlockchainError> {
         trace!(target: "backend", "creating new block with {} transactions", pool_transactions.len());
 
         let (outcome, header, block_hash) = {

crates/anvil/src/eth/backend/notifications.rs

@@ -1,6 +1,7 @@
 //! Notifications emitted from the backed
 
 use alloy_consensus::Header;
+use alloy_eips::eip7840::BlobParams;
 use alloy_primitives::B256;
 use alloy_rpc_types::Log;
 use futures::channel::mpsc::UnboundedReceiver;
@@ -37,5 +38,13 @@ pub struct NewBlockNotification {
     pub header: Arc<Header>,
 }
 
+#[derive(Clone, Debug)]
+pub(crate) struct FeeHistoryNotification {
+    pub(crate) hash: B256,
+    pub(crate) header: Arc<Header>,
+    pub(crate) generation: u64,
+    pub(crate) blob_params: BlobParams,
+}
+
 /// Type alias for a receiver that receives [ChainNotification]
 pub type ChainNotifications = UnboundedReceiver<ChainNotification>;

crates/anvil/src/eth/fees.rs

@@ -2,7 +2,10 @@ use std::{
     collections::BTreeMap,
     fmt,
     pin::Pin,
-    sync::{Arc, LazyLock},
+    sync::{
+        Arc, LazyLock,
+        atomic::{AtomicU64, Ordering},
+    },
     task::{Context, Poll},
 };
 
@@ -16,7 +19,7 @@ use revm::{context_interface::block::BlobExcessGasAndPrice, primitives::hardfork
 use tempo_hardfork::{TempoHardfork, constants::gas::tempo_t7_next_block_base_fee};
 
 use crate::eth::{
-    backend::{info::StorageInfo, notifications::ChainNotifications},
+    backend::{info::StorageInfo, notifications::FeeHistoryNotification},
     error::BlockchainError,
 };
 
@@ -280,10 +283,10 @@ pub struct FeeHistoryService<N: Network>
 where
     N::ReceiptEnvelope: TxReceipt<Log = alloy_primitives::Log>,
 {
-    /// Live fee rules, including blob parameters replaced by fork resets.
-    fees: FeeManager,
-    /// incoming notifications about new blocks
-    new_blocks: ChainNotifications,
+    /// Current reset generation.
+    generation: Arc<AtomicU64>,
+    /// New blocks with the fee rules active when each block was mined.
+    new_blocks: futures::channel::mpsc::UnboundedReceiver<FeeHistoryNotification>,
     /// contains all fee history related entries
     cache: FeeHistoryCache,
     /// number of items to consider
@@ -296,14 +299,14 @@ impl<N: Network> FeeHistoryService<N>
 where
     N::ReceiptEnvelope: TxReceipt<Log = alloy_primitives::Log>,
 {
-    pub const fn new(
-        fees: FeeManager,
-        new_blocks: ChainNotifications,
+    pub(crate) const fn new(
+        generation: Arc<AtomicU64>,
+        new_blocks: futures::channel::mpsc::UnboundedReceiver<FeeHistoryNotification>,
         cache: FeeHistoryCache,
         storage_info: StorageInfo<N>,
     ) -> Self {
         Self {
-            fees,
+            generation,
             new_blocks,
             cache,
             fee_history_limit: MAX_FEE_HISTORY_CACHE_SIZE,
@@ -317,45 +320,30 @@ where
     }
 
     /// Inserts a new cache entry for the given block
-    pub(crate) fn insert_cache_entry_for_block(&self, hash: B256, header: &impl BlockHeader) {
-        let (result, block_number) = self.create_cache_entry(hash, header);
-        self.insert_cache_entry(result, block_number);
-    }
-
-    /// Create a new history entry for the block
-    fn create_cache_entry(
+    pub(crate) fn insert_cache_entry_for_block(
         &self,
         hash: B256,
         header: &impl BlockHeader,
-    ) -> (FeeHistoryCacheItem, Option<u64>) {
-        create_fee_history_cache_item(hash, header, &self.storage_info, self.fees.blob_params())
-    }
-
-    fn insert_cache_entry(&self, item: FeeHistoryCacheItem, block_number: Option<u64>) {
-        insert_fee_history_cache_item(&self.cache, item, block_number, self.fee_history_limit);
+        blob_params: BlobParams,
+    ) {
+        let (item, block_number) =
+            create_fee_history_cache_item(hash, header, &self.storage_info, blob_params);
+        self.insert_cache_entry(item, block_number, self.generation.load(Ordering::Acquire));
     }
-}
 
-/// Inserts an entry into the fee history cache and trims it back to `fee_history_limit`.
-///
-/// Used by the async [`FeeHistoryService`]. The `eth_feeHistory` fallback applies the same bounded
-/// insertion policy to a batch under one lock.
-pub(crate) fn insert_fee_history_cache_item(
-    cache: &FeeHistoryCache,
-    item: FeeHistoryCacheItem,
-    block_number: Option<u64>,
-    fee_history_limit: u64,
-) {
-    if let Some(block_number) = block_number {
-        trace!(target: "fees", "insert new history item={:?} for {}", item, block_number);
-        let mut cache = cache.lock();
+    fn insert_cache_entry(
+        &self,
+        item: FeeHistoryCacheItem,
+        block_number: Option<u64>,
+        generation: u64,
+    ) {
+        let Some(block_number) = block_number else { return };
+        let mut cache = self.cache.lock();
+        if generation != self.generation.load(Ordering::Acquire) {
+            return;
+        }
         cache.insert(block_number, item);
-
-        // Trim to the cache limit by dropping the oldest entries (smallest block numbers).
-        // `pop_first` is saturating and correct regardless of insertion order, unlike the
-        // previous index math which could underflow when the `eth_feeHistory` fallback inserts
-        // entries out of order.
-        while cache.len() as u64 > fee_history_limit {
+        while cache.len() as u64 > self.fee_history_limit {
             cache.pop_first();
         }
     }
@@ -486,10 +474,16 @@ where
         let pin = self.get_mut();
 
         while let Poll::Ready(Some(notification)) = pin.new_blocks.poll_next_unpin(cx) {
-            // add the imported block.
-            if let Some(block) = notification.as_new_block() {
-                pin.insert_cache_entry_for_block(block.hash, block.header.as_ref());
+            if notification.generation != pin.generation.load(Ordering::Acquire) {
+                continue;
             }
+            let (item, block_number) = create_fee_history_cache_item(
+                notification.hash,
+                notification.header.as_ref(),
+                &pin.storage_info,
+                notification.blob_params,
+            );
+            pin.insert_cache_entry(item, block_number, notification.generation);
         }
 
         Poll::Pending