From 5d06758e7ef94b8ff56d67435ab34038dfd55133 Mon Sep 17 00:00:00 2001 From: Francis Li Date: Mon, 5 Oct 2026 19:17:36 +0000 Subject: [PATCH] fix(anvil): serialize state operations with mining Amp-Thread-ID: https://ampcode.com/threads/T-01a10d69-00b6-7378-855f-af2b1a148692 Co-authored-by: Amp --- crates/anvil/src/cmd.rs | 50 ++++++++++++++++++++++++- crates/anvil/src/eth/backend/mem/mod.rs | 7 ++++ 2 files changed, 56 insertions(+), 1 deletion(-) diff --git a/crates/anvil/src/cmd.rs b/crates/anvil/src/cmd.rs index 995829ae7..5407d8f16 100644 --- a/crates/anvil/src/cmd.rs +++ b/crates/anvil/src/cmd.rs @@ -650,7 +650,9 @@ impl PeriodicStateDumper { Self { in_progress_dump: None, api, dump_state, preserve_historical_states, interval } } - async fn dump(&self) { + async fn dump(mut self) { + // The suspended periodic dump may own the mining lock. Drop it before the final dump. + self.in_progress_dump = None; if let Some(state) = self.dump_state.clone() { Self::dump_state(self.api.clone(), state, self.preserve_historical_states).await } @@ -800,6 +802,52 @@ mod tests { use super::*; use std::{env, net::Ipv4Addr}; + #[tokio::test(flavor = "multi_thread")] + async fn shutdown_cancels_suspended_periodic_dump() { + let (api, _handle) = crate::spawn(NodeConfig::test()).await; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("state.json"); + let db_guard = api.backend.get_db().write().await; + let mut dumper = PeriodicStateDumper::new( + api.clone(), + Some(path.clone()), + Duration::from_secs(60), + false, + ); + dumper.in_progress_dump = + Some(Box::pin(PeriodicStateDumper::dump_state(api.clone(), path.clone(), false))); + // Suspend the periodic dump after taking the mining lock, while awaiting the database. + assert!(futures::poll!(&mut dumper).is_pending()); + drop(db_guard); + tokio::time::timeout(Duration::from_secs(30), dumper.dump()).await.unwrap(); + let saved = SerializableState::load(&path).unwrap(); + assert_eq!(saved.best_block_number, Some(api.backend.best_number())); + } + + #[tokio::test(flavor = "multi_thread")] + async fn state_operations_wait_for_in_progress_mining() { + let (api, _handle) = crate::spawn(NodeConfig::test()).await; + let db_guard = api.backend.get_db().write().await; + let mut mining = Box::pin(api.backend.mine_block(Vec::new())); + assert!(futures::poll!(&mut mining).is_pending()); + let mut dump = Box::pin(api.serialized_state(false)); + assert!(futures::poll!(&mut dump).is_pending()); + let mut snapshot = Box::pin(api.evm_snapshot()); + assert!(futures::poll!(&mut snapshot).is_pending()); + drop(db_guard); + let (mined, saved, snapshot) = tokio::time::timeout(Duration::from_secs(30), async { + tokio::join!(mining, dump, snapshot) + }) + .await + .unwrap(); + let saved = saved.unwrap(); + assert_eq!(saved.best_block_number, Some(mined.block_number)); + assert_eq!(saved.block.unwrap().number, U256::from(mined.block_number)); + assert_eq!(saved.blocks.len(), 2); + let snapshots = api.backend.list_state_snapshots(); + assert_eq!(snapshots[&snapshot.unwrap()].0, mined.block_number); + } + #[test] fn test_parse_fork_url() { let fork: ForkUrl = "http://localhost:8545@1000000".parse().unwrap(); diff --git a/crates/anvil/src/eth/backend/mem/mod.rs b/crates/anvil/src/eth/backend/mem/mod.rs index c29e4dd40..c2bb567f1 100644 --- a/crates/anvil/src/eth/backend/mem/mod.rs +++ b/crates/anvil/src/eth/backend/mem/mod.rs @@ -518,6 +518,7 @@ impl Backend { /// Resets the fork to a fresh state pub async fn reset_fork(&self, forking: Forking) -> Result<(), BlockchainError> { + let _mining_guard = self.mining.lock().await; if !self.is_fork() { if let Some(eth_rpc_url) = forking.clone().json_rpc_url { let mut env = self.env.read().clone(); @@ -624,6 +625,7 @@ impl Backend { /// Resets the backend to a fresh in-memory state, clearing all existing data pub async fn reset_to_in_mem(&self) -> Result<(), BlockchainError> { + let _mining_guard = self.mining.lock().await; // Clear the fork if any exists *self.fork.write() = None; @@ -969,6 +971,7 @@ impl Backend { /// /// Returns the id of the snapshot created. pub async fn create_state_snapshot(&self) -> U256 { + let _mining_guard = self.mining.lock().await; let num = self.best_number(); let hash = self.best_hash(); let id = self.db.write().await.snapshot_state(); @@ -979,6 +982,7 @@ impl Backend { /// Reverts the state to the state snapshot identified by the given `id`. pub async fn revert_state_snapshot(&self, id: U256) -> Result { + let _mining_guard = self.mining.lock().await; let block = { self.active_state_snapshots.lock().remove(&id) }; if let Some((num, hash)) = block { let best_block_hash = { @@ -1033,6 +1037,7 @@ impl Backend { &self, preserve_historical_states: bool, ) -> Result { + let _mining_guard = self.mining.lock().await; let at = self.env.read().evm_env.block_env.clone(); let best_number = self.blockchain.storage.read().best_number; let blocks = self.blockchain.storage.read().serialized_blocks(); @@ -1071,6 +1076,7 @@ impl Backend { /// Apply [SerializableState] data to the backend storage. pub async fn load_state(&self, state: SerializableState) -> Result { + let _mining_guard = self.mining.lock().await; // load the blocks and transactions into the storage self.blockchain.storage.write().load_blocks(state.blocks.clone()); self.blockchain.storage.write().load_transactions(state.transactions.clone()); @@ -3469,6 +3475,7 @@ impl Backend { /// The state of the chain is rewound using `rewind` to the common block, including the db, /// storage, and env. pub async fn rollback(&self, common_block: Block) -> Result<(), BlockchainError> { + let _mining_guard = self.mining.lock().await; // Get the database at the common block let common_state = { let return_state_or_throw_err =