Serialize state operations with mining

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();

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<bool, BlockchainError> {
+        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<SerializableState, BlockchainError> {
+        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<bool, BlockchainError> {
+        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 =