Repository: matter-labs/era-test-node (rust)

Issue:
feat: implement `anvil_dumpState`/`anvil_loadState` API endpoints
We can likely reuse snapshot logic for this. Ideally #414 should be done before this is worked on to avoid changes in serialization format (although likely we don't want to provide any cross-version guarantees here).

```rust
#[rpc(name = "anvil_dumpState")]
fn dump_state(&self, preserve_historical_states: Option<bool>) -> RpcResult<Bytes>;

#[rpc(name = "anvil_loadState")]
fn load_state(&self, bytes: Bytes) -> RpcResult<bool>;
```

Pull request:
feat: add `anvil_dumpState`/`anvil_loadState`
# What :computer: 
Closes #418 

Ditto. Required making `TransactionResult` serializable so unfortunately touches some seemingly unrelated VM stuff. Had to drop some non-serializable VM structs from state which honestly simplified logic a little bit.

# Why :hand:
Anvil feature parity

Patch:
diff --git a/Cargo.lock b/Cargo.lock
index b159b70..d20aa6e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -733,6 +733,7 @@ dependencies = [
  "ethabi 16.0.0",
  "ethers",
  "eyre",
+ "flate2",
  "fs2",
  "futures 0.3.31",
  "hex",
@@ -758,6 +759,7 @@ dependencies = [
  "tempdir",
  "test-case",
  "test-log",
+ "thiserror",
  "time",
  "tokio",
  "tracing",
diff --git a/Cargo.toml b/Cargo.toml
index 70189cd..df60129 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -59,6 +59,8 @@ indexmap = "2.0.1"
 chrono = { version = "0.4.31", default-features = false }
 time = "0.3.36"
 rand = "0.8"
+flate2 = "1.0"
+thiserror = "1"
 
 [dev-dependencies]
 httptest = "0.15.4"
diff --git a/src/config/cli.rs b/src/config/cli.rs
index c59448a..1ffee7c 100644
--- a/src/config/cli.rs
+++ b/src/config/cli.rs
@@ -52,7 +52,7 @@ pub struct Cli {
         long,
         value_name = "IP_ADDR",
         env = "ANVIL_ZKSYNC_IP_ADDR",
-        default_value = "127.0.0.1",
+        default_value = "0.0.0.0",
         value_delimiter = ',',
         help_heading = "Network Options"
     )]
@@ -444,7 +444,7 @@ mod tests {
     fn can_parse_host() {
         // Test adapted from https://github.com/foundry-rs/foundry/blob/398ef4a3d55d8dd769ce86cada5ec845e805188b/crates/anvil/src/cmd.rs#L895
         let args = Cli::parse_from(["anvil-zksync"]);
-        assert_eq!(args.host, vec![IpAddr::V4(Ipv4Addr::LOCALHOST)]);
+        assert_eq!(args.host, vec![IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0))]);
 
         let args = Cli::parse_from([
             "anvil-zksync",
diff --git a/src/fork.rs b/src/fork.rs
index fcf6c10..fcaf1ce 100644
--- a/src/fork.rs
+++ b/src/fork.rs
@@ -3,6 +3,19 @@
 //! There is ForkStorage (that is a wrapper over InMemoryStorage)
 //! And ForkDetails - that parses network address and fork height from arguments.
 
+use crate::config::{
+    cache::CacheConfig,
+    constants::{
+        DEFAULT_ESTIMATE_GAS_PRICE_SCALE_FACTOR, DEFAULT_ESTIMATE_GAS_SCALE_FACTOR,
+        DEFAULT_FAIR_PUBDATA_PRICE, TEST_NODE_NETWORK_ID,
+    },
+};
+use crate::system_contracts;
+use crate::{deps::InMemoryStorage, http_fork_source::HttpForkSource};
+use eyre::eyre;
+use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
+use std::iter::FromIterator;
 use std::{
     collections::HashMap,
     convert::{TryFrom, TryInto},
@@ -12,11 +25,9 @@ use std::{
     str::FromStr,
     sync::{Arc, RwLock},
 };
-
-use eyre::eyre;
 use tokio::runtime::Builder;
-use zksync_types::{Address, L1BatchNumber, L2BlockNumber, L2ChainId, H256, U256, U64};
-
+use zksync_multivm::interface::storage::ReadStorage;
+use zksync_types::web3::Bytes;
 use zksync_types::{
     api::{
         Block, BlockDetails, BlockIdVariant, BlockNumber, BridgeAddresses, Transaction,
@@ -27,27 +38,16 @@ use zksync_types::{
     url::SensitiveUrl,
     ProtocolVersionId, StorageKey,
 };
-
-use zksync_multivm::interface::storage::ReadStorage;
+use zksync_types::{
+    Address, L1BatchNumber, L2BlockNumber, L2ChainId, StorageValue, H256, U256, U64,
+};
 use zksync_utils::{bytecode::hash_bytecode, h256_to_u256};
-
 use zksync_web3_decl::{
     client::{Client, L2},
     namespaces::ZksNamespaceClient,
 };
 use zksync_web3_decl::{namespaces::EthNamespaceClient, types::Index};
 
-use crate::config::{
-    cache::CacheConfig,
-    constants::{
-        DEFAULT_ESTIMATE_GAS_PRICE_SCALE_FACTOR, DEFAULT_ESTIMATE_GAS_SCALE_FACTOR,
-        DEFAULT_FAIR_PUBDATA_PRICE, TEST_NODE_NETWORK_ID,
-    },
-};
-use crate::system_contracts;
-
-use crate::{deps::InMemoryStorage, http_fork_source::HttpForkSource};
-
 pub fn block_on<F: Future + Send + 'static>(future: F) -> F::Output
 where
     F::Output: Send,
@@ -262,6 +262,48 @@ impl<S: ForkSource> ForkStorage<S> {
         // TODO: Update this file to use proper enumeration index value once it's exposed for forks via API
         Some(0_u64)
     }
+
+    /// Creates a serializable representation of current storage state. It will contain both locally
+    /// stored data and cached data read from the fork.
+    pub fn dump_state(&self) -> SerializableForkStorage {
+        let inner = self.inner.read().unwrap();
+        let mut state = BTreeMap::from_iter(inner.value_read_cache.clone());
+        state.extend(inner.raw_storage.state.clone());
+        let mut factory_deps = BTreeMap::from_iter(
+            inner
+                .factory_dep_cache
+                .iter()
+                // Ignore cache misses
+                .filter_map(|(k, v)| v.as_ref().map(|v| (k, v)))
+                .map(|(k, v)| (*k, Bytes::from(v.clone()))),
+        );
+        factory_deps.extend(
+            inner
+                .raw_storage
+                .factory_deps
+                .iter()
+                .map(|(k, v)| (*k, Bytes::from(v.clone()))),
+        );
+
+        SerializableForkStorage {
+            storage: SerializableStorage(state),
+            factory_deps,
+        }
+    }
+
+    pub fn load_state(&self, state: SerializableForkStorage) {
+        tracing::trace!(
+            slots = state.storage.0.len(),
+            factory_deps = state.factory_deps.len(),
+            "loading fork storage from supplied state"
+        );
+        let mut inner = self.inner.write().unwrap();
+        inner.raw_storage.state.extend(state.storage.0);
+        inner
+            .raw_storage
+            .factory_deps
+            .extend(state.factory_deps.into_iter().map(|(k, v)| (k, v.0)));
+    }
 }
 
 impl<S: std::fmt::Debug + ForkSource> ReadStorage for ForkStorage<S> {
@@ -736,6 +778,85 @@ impl ForkDetails {
     }
 }
 
+/// Serializable representation of [`ForkStorage`]'s state.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct SerializableForkStorage {
+    /// Node's current key-value storage state (contains both local and cached fork data if applicable).
+    pub storage: SerializableStorage,
+    /// Factory dependencies by their hash.
+    pub factory_deps: BTreeMap<H256, Bytes>,
+}
+
+/// Wrapper for [`BTreeMap<StorageKey, StorageValue>`] to avoid serializing [`StorageKey`] as a struct.
+/// JSON does not support non-string keys so we use conversion to [`Bytes`] via [`crate::node::state::SerializableStorageKey`]
+/// instead.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(
+    into = "BTreeMap<serde_from::SerializableStorageKey, StorageValue>",
+    from = "BTreeMap<serde_from::SerializableStorageKey, StorageValue>"
+)]
+pub struct SerializableStorage(pub BTreeMap<StorageKey, StorageValue>);
+
+mod serde_from {
+    use crate::fork::SerializableStorage;
+    use serde::{Deserialize, Serialize};
+    use std::collections::BTreeMap;
+    use std::convert::TryFrom;
+    use zksync_types::web3::Bytes;
+    use zksync_types::{AccountTreeId, Address, StorageKey, StorageValue, H256};
+
+    impl From<BTreeMap<SerializableStorageKey, StorageValue>> for SerializableStorage {
+        fn from(value: BTreeMap<SerializableStorageKey, StorageValue>) -> Self {
+            SerializableStorage(value.into_iter().map(|(k, v)| (k.into(), v)).collect())
+        }
+    }
+
+    impl From<SerializableStorage> for BTreeMap<SerializableStorageKey, StorageValue> {
+        fn from(value: SerializableStorage) -> Self {
+            value.0.into_iter().map(|(k, v)| (k.into(), v)).collect()
+        }
+    }
+
+    #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
+    #[serde(into = "Bytes", try_from = "Bytes")]
+    pub struct SerializableStorageKey(StorageKey);
+
+    impl From<StorageKey> for SerializableStorageKey {
+        fn from(value: StorageKey) -> Self {
+            SerializableStorageKey(value)
+        }
+    }
+
+    impl From<SerializableStorageKey> for StorageKey {
+        fn from(value: SerializableStorageKey) -> Self {
+            value.0
+        }
+    }
+
+    impl TryFrom<Bytes> for SerializableStorageKey {
+        type Error = anyhow::Error;
+
+        fn try_from(bytes: Bytes) -> anyhow::Result<Self> {
+            if bytes.0.len() != 52 {
+                anyhow::bail!("invalid bytes length (expected 52, got {})", bytes.0.len())
+            }
+            let address = Address::from_slice(&bytes.0[0..20]);
+            let key = H256::from_slice(&bytes.0[20..52]);
+            Ok(SerializableStorageKey(StorageKey::new(
+                AccountTreeId::new(address),
+                key,
+            )))
+        }
+    }
+
+    impl From<SerializableStorageKey> for Bytes {
+        fn from(value: SerializableStorageKey) -> Self {
+            let bytes = [value.0.address().as_bytes(), value.0.key().as_bytes()].concat();
+            bytes.into()
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use zksync_multivm::interface::storage::ReadStorage;
diff --git a/src/namespaces/anvil.rs b/src/namespaces/anvil.rs
index 7efeff2..2a76e67 100644
--- a/src/namespaces/anvil.rs
+++ b/src/namespaces/anvil.rs
@@ -8,6 +8,30 @@ use zksync_types::{Address, H256, U256, U64};
 
 #[rpc]
 pub trait AnvilNamespaceT {
+    /// Create a buffer that represents all state on the chain, which can be loaded to separate
+    /// process by calling `anvil_loadState`.
+    ///
+    /// # Arguments
+    ///
+    /// * `preserve_historical_states` - Whether to preserve historical states
+    ///
+    /// # Returns
+    /// Buffer representing the chain state.
+    #[rpc(name = "anvil_dumpState")]
+    fn dump_state(&self, preserve_historical_states: Option<bool>) -> RpcResult<Bytes>;
+
+    /// Append chain state buffer to current chain. Will overwrite any conflicting addresses or
+    /// storage.
+    ///
+    /// # Arguments
+    ///
+    /// * `bytes` - Buffer containing the chain state
+    ///
+    /// # Returns
+    /// `true` if a snapshot was reverted, otherwise `false`.
+    #[rpc(name = "anvil_loadState")]
+    fn load_state(&self, bytes: Bytes) -> RpcResult<bool>;
+
     /// Mines a single block in the same way as `evm_mine` but returns extra fields.
     ///
     /// # Returns
diff --git a/src/node/anvil.rs b/src/node/anvil.rs
index 7c7141f..133375c 100644
--- a/src/node/anvil.rs
+++ b/src/node/anvil.rs
@@ -1,4 +1,5 @@
 use zksync_types::api::Block;
+use zksync_types::web3::Bytes;
 use zksync_types::{Address, H256, U256, U64};
 use zksync_web3_decl::error::Web3Error;
 
@@ -14,6 +15,21 @@ use crate::{
 impl<S: ForkSource + std::fmt::Debug + Clone + Send + Sync + 'static> AnvilNamespaceT
     for InMemoryNode<S>
 {
+    fn dump_state(&self, preserve_historical_states: Option<bool>) -> RpcResult<Bytes> {
+        self.dump_state(preserve_historical_states.unwrap_or(false))
+            .map_err(|err| {
+                tracing::error!("failed dumping state: {:?}", err);
+                into_jsrpc_error(Web3Error::InternalError(err))
+            })
+            .into_boxed_future()
+    }
+
+    fn load_state(&self, bytes: Bytes) -> RpcResult<bool> {
+        self.load_state(bytes)
+            .map_err(Into::into)
+            .into_boxed_future()
+    }
+
     fn mine_detailed(&self) -> RpcResult<Block<DetailedTransaction>> {
         self.mine_detailed()
             .map_err(|err| {
diff --git a/src/node/error.rs b/src/node/error.rs
new file mode 100644
index 0000000..745ae15
--- /dev/null
+++ b/src/node/error.rs
@@ -0,0 +1,36 @@
+use crate::utils::into_jsrpc_error;
+use zksync_web3_decl::error::Web3Error;
+
+#[derive(thiserror::Error, Debug)]
+pub enum LoadStateError {
+    #[error("loading state into a node with existing state is not allowed (please create an issue if you have a valid use case)")]
+    HasExistingState,
+    #[error("loading empty state (no blocks) is not allowed")]
+    EmptyState,
+    #[error("failed to decompress state: {0}")]
+    FailedDecompress(std::io::Error),
+    #[error("failed to deserialize state: {0}")]
+    FailedDeserialize(serde_json::Error),
+    #[error("unknown state version `{0}`")]
+    UnknownStateVersion(u8),
+    #[error(transparent)]
+    Other(#[from] anyhow::Error),
+}
+
+impl From<LoadStateError> for jsonrpc_core::Error {
+    fn from(value: LoadStateError) -> Self {
+        match value {
+            err @ LoadStateError::HasExistingState
+            | err @ LoadStateError::EmptyState
+            | err @ LoadStateError::FailedDecompress(_)
+            | err @ LoadStateError::FailedDeserialize(_)
+            | err @ LoadStateError::UnknownStateVersion(_) => {
+                jsonrpc_core::Error::invalid_params(err.to_string())
+            }
+            LoadStateError::Other(err) => {
+                tracing::error!("failed loading state: {:?}", err);
+                into_jsrpc_error(Web3Error::InternalError(err))
+            }
+        }
+    }
+}
diff --git a/src/node/in_memory.rs b/src/node/in_memory.rs
index f57a270..8b15321 100644
--- a/src/node/in_memory.rs
+++ b/src/node/in_memory.rs
@@ -1,7 +1,12 @@
 //! In-memory node, that supports forking other networks.
 use colored::Colorize;
+use flate2::read::GzDecoder;
+use flate2::write::GzEncoder;
+use flate2::Compression;
 use indexmap::IndexMap;
 use once_cell::sync::OnceCell;
+use serde::{Deserialize, Serialize};
+use std::io::{Read, Write};
 use std::sync::{RwLockReadGuard, RwLockWriteGuard};
 use std::{
     collections::{HashMap, HashSet},
@@ -9,7 +14,6 @@ use std::{
     str::FromStr,
     sync::{Arc, RwLock},
 };
-
 use zksync_contracts::BaseSystemContracts;
 use zksync_multivm::vm_latest::HistoryEnabled;
 use zksync_multivm::{
@@ -48,7 +52,10 @@ use zksync_types::{
 use zksync_utils::{bytecode::hash_bytecode, h256_to_account_address, h256_to_u256, u256_to_h256};
 use zksync_web3_decl::error::Web3Error;
 
+use crate::fork::SerializableStorage;
+use crate::node::error::LoadStateError;
 use crate::node::impersonate::{ImpersonationManager, ImpersonationState};
+use crate::node::state::{StateV1, VersionedState};
 use crate::node::time::{AdvanceTime, ReadTime, TimestampManager};
 use crate::node::{BlockSealer, TxPool};
 use crate::{
@@ -196,17 +203,15 @@ fn create_block<TX>(
 }
 
 /// Information about the executed transaction.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct TxExecutionInfo {
     pub tx: L2Tx,
     // Batch number where transaction was executed.
     pub batch_number: u32,
     pub miniblock_number: u64,
-    #[allow(unused)]
-    pub result: VmExecutionResultAndLogs,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct TransactionResult {
     pub info: TxExecutionInfo,
     pub receipt: TransactionReceipt,
@@ -876,6 +881,119 @@ impl<S: std::fmt::Debug + ForkSource> InMemoryNodeInner<S> {
         Ok(())
     }
 
+    fn dump_state(&self, preserve_historical_states: bool) -> anyhow::Result<VersionedState> {
+        let fork_storage = self.fork_storage.dump_state();
+        let historical_states = if preserve_historical_states {
+            self.previous_states
+                .iter()
+                .map(|(k, v)| (*k, SerializableStorage(v.clone().into_iter().collect())))
+                .collect()
+        } else {
+            Vec::new()
+        };
+
+        Ok(VersionedState::v1(StateV1 {
+            blocks: self.blocks.values().cloned().collect(),
+            transactions: self.tx_results.values().cloned().collect(),
+            fork_storage,
+            historical_states,
+        }))
+    }
+
+    fn load_blocks<T: AdvanceTime>(&mut self, mut time: T, blocks: Vec<Block<TransactionVariant>>) {
+        tracing::trace!(
+            blocks = blocks.len(),
+            "loading new blocks from supplied state"
+        );
+        for block in blocks {
+            let number = block.number.as_u64();
+            tracing::trace!(
+                number,
+                hash = %block.hash,
+                "loading new block from supplied state"
+            );
+
+            self.block_hashes.insert(number, block.hash);
+            self.blocks.insert(block.hash, block);
+        }
+
+        // Safe unwrap as there was at least one block in the loaded state
+        let latest_block = self.blocks.values().max_by_key(|b| b.number).unwrap();
+        let latest_number = latest_block.number.as_u64();
+        let latest_hash = latest_block.hash;
+        let Some(latest_batch_number) = latest_block.l1_batch_number.map(|n| n.as_u32()) else {
+            panic!("encountered a block with no batch; this is not supposed to happen")
+        };
+        let latest_timestamp = latest_block.timestamp.as_u64();
+        tracing::info!(
+            number = latest_number,
+            hash = %latest_hash,
+            batch_number = latest_batch_number,
+            timestamp = latest_timestamp,
+            "latest block after loading state"
+        );
+        self.current_miniblock = latest_number;
+        self.current_miniblock_hash = latest_hash;
+        self.current_batch = latest_batch_number;
+        time.reset_to(latest_timestamp);
+    }
+
+    fn load_transactions(&mut self, transactions: Vec<TransactionResult>) {
+        tracing::trace!(
+            transactions = transactions.len(),
+            "loading new transactions from supplied state"
+        );
+        for transaction in transactions {
+            tracing::trace!(
+                hash = %transaction.receipt.transaction_hash,
+                "loading new transaction from supplied state"
+            );
+            self.tx_results
+                .insert(transaction.receipt.transaction_hash, transaction);
+        }
+    }
+
+    fn load_state<T: AdvanceTime>(
+        &mut self,
+        time: T,
+        state: VersionedState,
+    ) -> Result<bool, LoadStateError> {
+        if self.blocks.len() > 1 {
+            tracing::debug!(
+                blocks = self.blocks.len(),
+                "node has existing state; refusing to load new state"
+            );
+            return Err(LoadStateError::HasExistingState);
+        }
+        let state = match state {
+            VersionedState::V1 { state, .. } => state,
+            VersionedState::Unknown { version } => {
+                return Err(LoadStateError::UnknownStateVersion(version))
+            }
+        };
+        if state.blocks.is_empty() {
+            tracing::debug!("new state has no blocks; refusing to load");
+            return Err(LoadStateError::EmptyState);
+        }
+
+        self.load_blocks(time, state.blocks);
+        self.load_transactions(state.transactions);
+        self.fork_storage.load_state(state.fork_storage);
+
+        tracing::trace!(
+            states = state.historical_states.len(),
+            "loading historical states from supplied state"
+        );
+        self.previous_states.extend(
+            state
+                .historical_states
+                .into_iter()
+                .map(|(k, v)| (k, v.0.into_iter().collect())),
+        );
+
+        Ok(true)
+    }
+
     fn apply_block<T: AdvanceTime>(
         &mut self,
         time: &mut T,
@@ -1652,7 +1770,6 @@ impl<S: ForkSource + std::fmt::Debug + Clone> InMemoryNode<S> {
                     tx: l2_tx,
                     batch_number: batch_env.number.0,
                     miniblock_number: block_ctx.miniblock,
-                    result,
                 },
                 receipt: tx_receipt,
                 debug,
@@ -1716,13 +1833,15 @@ impl<S: ForkSource + std::fmt::Debug + Clone> InMemoryNode<S> {
         }
 
         let mut transactions = Vec::new();
-        let mut tx_results = Vec::new();
+        let mut tx_receipts = Vec::new();
+        let mut debug_calls = Vec::new();
         for tx_hash in &executed_tx_hashes {
             let Some(tx_result) = inner.tx_results.get(tx_hash) else {
                 // Skipping halted transaction
                 continue;
             };
-            tx_results.push(&tx_result.info.result);
+            tx_receipts.push(&tx_result.receipt);
+            debug_calls.push(&tx_result.debug);
 
             let mut transaction = zksync_types::api::Transaction::from(tx_result.info.tx.clone());
             transaction.block_hash = Some(block_ctx.hash);
@@ -1741,12 +1860,12 @@ impl<S: ForkSource + std::fmt::Debug + Clone> InMemoryNode<S> {
         }
 
         // Build bloom hash
-        let iter = tx_results
+        let iter = tx_receipts
             .iter()
-            .flat_map(|r| r.logs.events.iter())
+            .flat_map(|r| r.logs.iter())
             .flat_map(|event| {
                 event
-                    .indexed_topics
+                    .topics
                     .iter()
                     .map(|topic| BloomInput::Raw(topic.as_bytes()))
                     .chain([BloomInput::Raw(event.address.as_bytes())])
@@ -1754,9 +1873,9 @@ impl<S: ForkSource + std::fmt::Debug + Clone> InMemoryNode<S> {
         let logs_bloom = build_bloom(iter);
 
         // Calculate how much gas was used across all txs
-        let gas_used = tx_results
+        let gas_used = debug_calls
             .iter()
-            .map(|r| U256::from(r.statistics.gas_used))
+            .map(|r| r.gas_used)
             .fold(U256::zero(), |acc, x| acc + x);
 
         // Construct the block
@@ -1824,6 +1943,43 @@ impl<S: ForkSource + std::fmt::Debug + Clone> InMemoryNode<S> {
 
         Ok(())
     }
+
+    pub fn dump_state(&self, preserve_historical_states: bool) -> anyhow::Result<Bytes> {
+        let state = self
+            .inner
+            .read()
+            .map_err(|_| anyhow::anyhow!("Failed to acquire read lock"))?
+            .dump_state(preserve_historical_states)?;
+        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
+        encoder.write_all(&serde_json::to_vec(&state)?)?;
+        Ok(encoder.finish()?.into())
+    }
+
+    pub fn load_state(&self, buf: Bytes) -> Result<bool, LoadStateError> {
+        let orig_buf = &buf.0[..];
+        let mut decoder = GzDecoder::new(orig_buf);
+        let mut decoded_data = Vec::new();
+
+        // Support both compressed and non-compressed state format
+        let decoded = if decoder.header().is_some() {
+            tracing::trace!(bytes = buf.0.len(), "decompressing state");
+            decoder
+                .read_to_end(decoded_data.as_mut())
+                .map_err(LoadStateError::FailedDecompress)?;
+            &decoded_data
+        } else {
+            &buf.0
+        };
+        tracing::trace!(bytes = decoded.len(), "deserializing state");
+        let state: VersionedState =
+            serde_json::from_slice(decoded).map_err(LoadStateError::FailedDeserialize)?;
+
+        let time = self.time.lock();
+        self.inner
+            .write()
+            .map_err(|_| anyhow::anyhow!("Failed to acquire write lock"))?
+            .load_state(time, state)
+    }
 }
 
 /// Keeps track of a block's batch number, miniblock number and timestamp.
diff --git a/src/node/in_memory_ext.rs b/src/node/in_memory_ext.rs
index e7e66f1..0d89cb9 100644
--- a/src/node/in_memory_ext.rs
+++ b/src/node/in_memory_ext.rs
@@ -11,7 +11,7 @@ use crate::{
 use anyhow::{anyhow, Context};
 use std::convert::TryInto;
 use std::time::Duration;
-use zksync_multivm::interface::{ExecutionResult, TxExecutionMode};
+use zksync_multivm::interface::TxExecutionMode;
 use zksync_types::api::{Block, TransactionVariant};
 use zksync_types::{
     get_code_key, get_nonce_key,
@@ -111,15 +111,8 @@ impl<S: ForkSource + std::fmt::Debug + Clone + Send + Sync + 'static> InMemoryNo
                         .tx_results
                         .get(&tx.hash)
                         .expect("freshly executed tx is missing from storage");
-                    let (output, revert_reason) = match &tx_result.info.result.result {
-                        ExecutionResult::Success { output } => (Some(output.clone().into()), None),
-                        ExecutionResult::Revert { output } => (
-                            Some(output.encoded_data().into()),
-                            Some(output.to_user_friendly_string()),
-                        ),
-                        // Halted transaction should never be a part of a block
-                        ExecutionResult::Halt { .. } => unreachable!(),
-                    };
+                    let output = Some(tx_result.debug.output.clone());
+                    let revert_reason = tx_result.debug.revert_reason.clone();
                     DetailedTransaction {
                         inner: tx,
                         output,
diff --git a/src/node/mod.rs b/src/node/mod.rs
index 10ec112..0732e9e 100644
--- a/src/node/mod.rs
+++ b/src/node/mod.rs
@@ -5,6 +5,7 @@ mod block_producer;
 mod call_error_tracer;
 mod config_api;
 mod debug;
+mod error;
 mod eth;
 mod evm;
 mod fee_model;
@@ -15,6 +16,7 @@ mod in_memory_ext;
 mod net;
 mod pool;
 mod sealer;
+mod state;
 mod storage_logs;
 mod time;
 mod web3;
diff --git a/src/node/state.rs b/src/node/state.rs
new file mode 100644
index 0000000..a88fa2f
--- /dev/null
+++ b/src/node/state.rs
@@ -0,0 +1,67 @@
+use crate::fork::{SerializableForkStorage, SerializableStorage};
+use crate::node::TransactionResult;
+use serde::{Deserialize, Serialize};
+use zksync_types::api::{Block, TransactionVariant};
+use zksync_types::H256;
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum VersionedState {
+    V1 {
+        version: StateVersion<1>,
+        #[serde(flatten)]
+        state: StateV1,
+    },
+    Unknown {
+        version: u8,
+    },
+}
+
+impl VersionedState {
+    pub fn v1(state: StateV1) -> Self {
+        VersionedState::V1 {
+            version: StateVersion::<1>,
+            state,
+        }
+    }
+}
+
+/// Workaround while serde does not allow integer tags in enums (see https://github.com/serde-rs/serde/issues/745).
+#[derive(Copy, Clone, Debug)]
+pub struct StateVersion<const V: u8>;
+
+impl<const V: u8> Serialize for StateVersion<V> {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        serializer.serialize_u8(V)
+    }
+}
+
+impl<'de, const V: u8> Deserialize<'de> for StateVersion<V> {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let value = u8::deserialize(deserializer)?;
+        if value == V {
+            Ok(StateVersion::<V>)
+        } else {
+            Err(serde::de::Error::custom("unknown state version"))
+        }
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct StateV1 {
+    /// All blocks sealed on this node up to the current moment.
+    pub blocks: Vec<Block<TransactionVariant>>,
+    /// All transactions executed on this node up to the current moment.
+    pub transactions: Vec<TransactionResult>,
+    /// Current node's storage state.
+    #[serde(flatten)]
+    pub fork_storage: SerializableForkStorage,
+    /// Historical states of storage at particular block hashes.
+    pub historical_states: Vec<(H256, SerializableStorage)>,
+}
diff --git a/src/node/time.rs b/src/node/time.rs
index 379a6e0..0c5bac9 100644
--- a/src/node/time.rs
+++ b/src/node/time.rs
@@ -19,6 +19,8 @@ pub trait AdvanceTime: ReadTime {
     /// Subsequent calls to this method return monotonically increasing values. Time difference
     /// between calls is implementation-specific.
     fn advance_timestamp(&mut self) -> u64;
+
+    fn reset_to(&mut self, timestamp: u64);
 }
 
 /// Manages timestamps (in seconds) across the system.
@@ -153,11 +155,6 @@ struct TimestampManagerInternal {
 }
 
 impl TimestampManagerInternal {
-    fn reset_to(&mut self, timestamp: u64) {
-        self.next_timestamp.take();
-        self.current_timestamp = timestamp;
-    }
-
     fn interval(&self) -> u64 {
         self.interval.unwrap_or(1)
     }
@@ -184,6 +181,11 @@ impl AdvanceTime for TimestampManagerInternal {
         self.current_timestamp = next_timestamp;
         next_timestamp
     }
+
+    fn reset_to(&mut self, timestamp: u64) {
+        self.next_timestamp.take();
+        self.current_timestamp = timestamp;
+    }
 }
 
 struct TimeLockWithOffsets<'a> {
@@ -222,4 +224,12 @@ impl AdvanceTime for TimeLockWithOffsets<'_> {
             None => self.guard.advance_timestamp(),
         }
     }
+
+    fn reset_to(&mut self, timestamp: u64) {
+        // Resetting `start_timestamp` to `timestamp` may look weird here but at the same time there
+        // is no "expected" behavior in this case.
+        // Also, this is temporary logic that will become irrelevant after block production refactoring.
+        self.guard.reset_to(timestamp);
+        self.start_timestamp = timestamp;
+    }
 }
diff --git a/src/utils.rs b/src/utils.rs
index 8ff0032..9592ef1 100644
--- a/src/utils.rs
+++ b/src/utils.rs
@@ -132,7 +132,7 @@ pub fn create_debug_output(
         }),
         ExecutionResult::Revert { output } => Ok(DebugCall {
             gas_used: result.statistics.gas_used.into(),
-            output: Default::default(),
+            output: output.encoded_data().into(),
             r#type: calltype,
             from: l2_tx.initiator_account(),
             to: l2_tx.recipient_account().unwrap_or_default(),


Test patch:
diff --git a/e2e-tests-rust/Cargo.lock b/e2e-tests-rust/Cargo.lock
index a3437af..eb2e628 100644
--- a/e2e-tests-rust/Cargo.lock
+++ b/e2e-tests-rust/Cargo.lock
@@ -642,7 +642,7 @@ dependencies = [
 [[package]]
 name = "alloy-zksync"
 version = "0.6.1"
-source = "git+https://github.com/itegulov/alloy-zksync.git?rev=c43bba1a6c5e744afb975b261cba6e964d6a58c6#c43bba1a6c5e744afb975b261cba6e964d6a58c6"
+source = "git+https://github.com/itegulov/alloy-zksync.git?rev=692c5c2ca5defc88ac542f420d97c6756dadf9df#692c5c2ca5defc88ac542f420d97c6756dadf9df"
 dependencies = [
  "alloy",
  "async-trait",
diff --git a/e2e-tests-rust/Cargo.toml b/e2e-tests-rust/Cargo.toml
index 444d4c8..03503dd 100644
--- a/e2e-tests-rust/Cargo.toml
+++ b/e2e-tests-rust/Cargo.toml
@@ -10,7 +10,7 @@ categories = ["cryptography"]
 publish = false
 
 [dependencies]
-alloy-zksync = { git = "https://github.com/itegulov/alloy-zksync.git", rev = "c43bba1a6c5e744afb975b261cba6e964d6a58c6" }
+alloy-zksync = { git = "https://github.com/itegulov/alloy-zksync.git", rev = "692c5c2ca5defc88ac542f420d97c6756dadf9df" }
 alloy = { version = "0.6", features = ["full", "rlp", "serde", "sol-types", "getrandom", "provider-anvil-api"] }
 anyhow = "1.0"
 fs2 = "0.4.3"
diff --git a/e2e-tests-rust/src/ext.rs b/e2e-tests-rust/src/ext.rs
index 8b08f7e..7114a6d 100644
--- a/e2e-tests-rust/src/ext.rs
+++ b/e2e-tests-rust/src/ext.rs
@@ -26,6 +26,15 @@ pub trait ReceiptExt: ReceiptResponse {
         })
     }
 
+    fn sender(&self) -> anyhow::Result<Address> {
+        self.to().ok_or_else(|| {
+            anyhow::anyhow!(
+                "receipt (hash={}) does not have `to` address",
+                self.transaction_hash()
+            )
+        })
+    }
+
     /// Asserts that receipts belong to a block and that block is the same for both of them.
     fn assert_same_block(&self, other: &Self) -> anyhow::Result<()> {
         let lhs_number = self.block_number_ext()?;
@@ -45,6 +54,7 @@ pub trait ReceiptExt: ReceiptResponse {
             )
         }
     }
+
     /// Asserts that receipt is successful.
     fn assert_successful(&self) -> anyhow::Result<()> {
         if !self.status() {
diff --git a/e2e-tests-rust/src/lib.rs b/e2e-tests-rust/src/lib.rs
index 072c303..59a34eb 100644
--- a/e2e-tests-rust/src/lib.rs
+++ b/e2e-tests-rust/src/lib.rs
@@ -5,4 +5,4 @@ mod provider;
 mod utils;
 
 pub use ext::{ReceiptExt, ZksyncWalletProviderExt};
-pub use provider::{init_testing_provider, AnvilZKsyncApi, TestingProvider};
+pub use provider::{init_testing_provider, AnvilZKsyncApi, TestingProvider, DEFAULT_TX_VALUE};
diff --git a/e2e-tests-rust/src/provider/mod.rs b/e2e-tests-rust/src/provider/mod.rs
index e13184f..cf6e0b6 100644
--- a/e2e-tests-rust/src/provider/mod.rs
+++ b/e2e-tests-rust/src/provider/mod.rs
@@ -2,4 +2,4 @@ mod anvil_zksync;
 mod testing;
 
 pub use anvil_zksync::AnvilZKsyncApi;
-pub use testing::{init_testing_provider, TestingProvider};
+pub use testing::{init_testing_provider, TestingProvider, DEFAULT_TX_VALUE};
diff --git a/e2e-tests-rust/src/provider/testing.rs b/e2e-tests-rust/src/provider/testing.rs
index 1b83f64..dc5635d 100644
--- a/e2e-tests-rust/src/provider/testing.rs
+++ b/e2e-tests-rust/src/provider/testing.rs
@@ -1,19 +1,23 @@
 use crate::utils::LockedPort;
 use crate::ReceiptExt;
-use alloy::network::{Network, TransactionBuilder};
+use alloy::network::primitives::{BlockTransactionsKind, HeaderResponse as _};
+use alloy::network::{Network, ReceiptResponse as _, TransactionBuilder};
 use alloy::primitives::{Address, U256};
 use alloy::providers::{
     PendingTransaction, PendingTransactionBuilder, PendingTransactionError, Provider, RootProvider,
     SendableTx, WalletProvider,
 };
-use alloy::rpc::types::TransactionRequest;
+use alloy::rpc::types::{Block, TransactionRequest};
 use alloy::transports::http::{reqwest, Http};
 use alloy::transports::{RpcError, Transport, TransportErrorKind, TransportResult};
+use alloy_zksync::network::header_response::HeaderResponse;
 use alloy_zksync::network::receipt_response::ReceiptResponse;
+use alloy_zksync::network::transaction_response::TransactionResponse;
 use alloy_zksync::network::Zksync;
 use alloy_zksync::node_bindings::EraTestNode;
 use alloy_zksync::provider::{zksync_provider, ProviderBuilderExt};
 use alloy_zksync::wallet::ZksyncWallet;
+use anyhow::Context as _;
 use itertools::Itertools;
 use std::future::Future;
 use std::marker::PhantomData;
@@ -23,6 +27,8 @@ use std::task::{Context, Poll};
 use std::time::Duration;
 use tokio::task::JoinHandle;
 
+pub const DEFAULT_TX_VALUE: u64 = 100;
+
 /// Full requirements for the underlying Zksync provider.
 pub trait FullZksyncProvider<T>:
     Provider<T, Zksync> + WalletProvider<Zksync, Wallet = ZksyncWallet> + Clone
@@ -114,7 +120,7 @@ where
     pub fn tx(&self) -> TestTxBuilder<P, T> {
         let tx = TransactionRequest::default()
             .with_to(Address::random())
-            .with_value(U256::from(100));
+            .with_value(U256::from(DEFAULT_TX_VALUE));
         TestTxBuilder {
             inner: tx,
             provider: (*self).clone(),
@@ -161,6 +167,156 @@ where
     ) -> Result<RacedReceipts<N>, PendingTransactionError> {
         self.race_n_txs(|i, tx| tx.with_rich_from(i)).await
     }
+
+    pub async fn get_block_by_receipt(
+        &self,
+        receipt: &ReceiptResponse,
+    ) -> anyhow::Result<Block<TransactionResponse, HeaderResponse>> {
+        let hash = receipt.block_hash_ext()?;
+        self.get_block_by_hash(receipt.block_hash_ext()?, BlockTransactionsKind::Full)
+            .await?
+            .with_context(|| format!("block (hash={}) not found", hash))
+    }
+
+    pub async fn get_blocks_by_receipts(
+        &self,
+        receipts: impl IntoIterator<Item = &ReceiptResponse>,
+    ) -> anyhow::Result<Vec<Block<TransactionResponse, HeaderResponse>>> {
+        futures::future::join_all(
+            receipts
+                .into_iter()
+                .map(|receipt| self.get_block_by_receipt(receipt)),
+        )
+        .await
+        .into_iter()
+        .collect()
+    }
+
+    pub async fn assert_has_receipt(
+        &self,
+        expected_receipt: &ReceiptResponse,
+    ) -> anyhow::Result<()> {
+        let Some(actual_receipt) = self
+            .get_transaction_receipt(expected_receipt.transaction_hash())
+            .await?
+        else {
+            anyhow::bail!(
+                "receipt (hash={}) not found",
+                expected_receipt.transaction_hash()
+            );
+        };
+        assert_eq!(expected_receipt, &actual_receipt);
+        Ok(())
+    }
+
+    pub async fn assert_has_receipts(
+        &self,
+        receipts: impl IntoIterator<Item = &ReceiptResponse>,
+    ) -> anyhow::Result<()> {
+        for receipt in receipts {
+            self.assert_has_receipt(receipt).await?;
+        }
+        Ok(())
+    }
+
+    pub async fn assert_no_receipt(
+        &self,
+        expected_receipt: &ReceiptResponse,
+    ) -> anyhow::Result<()> {
+        if let Some(actual_receipt) = self
+            .get_transaction_receipt(expected_receipt.transaction_hash())
+            .await?
+        {
+            anyhow::bail!(
+                "receipt (hash={}) expected to be missing but was found",
+                actual_receipt.transaction_hash()
+            );
+        } else {
+            Ok(())
+        }
+    }
+
+    pub async fn assert_no_receipts(
+        &self,
+        receipts: impl IntoIterator<Item = &ReceiptResponse>,
+    ) -> anyhow::Result<()> {
+        for receipt in receipts {
+            self.assert_no_receipt(receipt).await?;
+        }
+        Ok(())
+    }
+
+    pub async fn assert_has_block(
+        &self,
+        expected_block: &Block<TransactionResponse, HeaderResponse>,
+    ) -> anyhow::Result<()> {
+        anyhow::ensure!(
+            expected_block.transactions.is_full(),
+            "expected block did not have full transactions"
+        );
+        let Some(actual_block) = self
+            .get_block_by_hash(expected_block.header.hash(), BlockTransactionsKind::Full)
+            .await?
+        else {
+            anyhow::bail!("block (hash={}) not found", expected_block.header.hash());
+        };
+        assert_eq!(expected_block, &actual_block);
+        Ok(())
+    }
+
+    pub async fn assert_has_blocks(
+        &self,
+        blocks: impl IntoIterator<Item = &Block<TransactionResponse, HeaderResponse>>,
+    ) -> anyhow::Result<()> {
+        for block in blocks {
+            self.assert_has_block(block).await?;
+        }
+        Ok(())
+    }
+
+    pub async fn assert_no_block(
+        &self,
+        expected_block: &Block<TransactionResponse, HeaderResponse>,
+    ) -> anyhow::Result<()> {
+        if let Some(actual_block) = self
+            .get_block_by_hash(expected_block.header.hash(), BlockTransactionsKind::Full)
+            .await?
+        {
+            anyhow::bail!(
+                "block (hash={}) expected to be missing but was found",
+                actual_block.header.hash()
+            );
+        } else {
+            Ok(())
+        }
+    }
+
+    pub async fn assert_no_blocks(
+        &self,
+        blocks: impl IntoIterator<Item = &Block<TransactionResponse, HeaderResponse>>,
+    ) -> anyhow::Result<()> {
+        for block in blocks {
+            self.assert_no_block(block).await?;
+        }
+        Ok(())
+    }
+
+    pub async fn assert_balance(
+        &self,
+        address: Address,
+        expected_balance: u64,
+    ) -> anyhow::Result<()> {
+        let actual_balance = self.get_balance(address).await?;
+        let expected_balance = U256::from(expected_balance);
+        anyhow::ensure!(
+            actual_balance == expected_balance,
+            "account's ({}) balance ({}) did not match expected value ({})",
+            address,
+            actual_balance,
+            expected_balance,
+        );
+        Ok(())
+    }
 }
 
 #[async_trait::async_trait]
diff --git a/e2e-tests-rust/tests/lib.rs b/e2e-tests-rust/tests/lib.rs
index e397a52..57b2aab 100644
--- a/e2e-tests-rust/tests/lib.rs
+++ b/e2e-tests-rust/tests/lib.rs
@@ -1,8 +1,9 @@
 use alloy::network::ReceiptResponse;
 use alloy::providers::ext::AnvilApi;
 use anvil_zksync_e2e_tests::{
-    init_testing_provider, AnvilZKsyncApi, ReceiptExt, ZksyncWalletProviderExt,
+    init_testing_provider, AnvilZKsyncApi, ReceiptExt, ZksyncWalletProviderExt, DEFAULT_TX_VALUE,
 };
+use std::convert::identity;
 use std::time::Duration;
 
 #[tokio::test]
@@ -252,3 +253,83 @@ async fn seal_block_ignoring_halted_transaction() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+#[tokio::test]
+async fn dump_and_load_state() -> anyhow::Result<()> {
+    // Test that we can submit transactions, then dump state and shutdown the node. Following that we
+    // should be able to spin up a new node and load state into it. Previous transactions/block should
+    // be present on the new node along with the old state.
+    let provider = init_testing_provider(identity).await?;
+
+    let receipts = [
+        provider.tx().finalize().await?,
+        provider.tx().finalize().await?,
+    ];
+    let blocks = provider.get_blocks_by_receipts(&receipts).await?;
+
+    // Dump node's state, re-create it and load old state
+    let state = provider.anvil_dump_state().await?;
+    let provider = init_testing_provider(identity).await?;
+    provider.anvil_load_state(state).await?;
+
+    // Assert that new node has pre-restart receipts, blocks and state
+    provider.assert_has_receipts(&receipts).await?;
+    provider.assert_has_blocks(&blocks).await?;
+    provider
+        .assert_balance(receipts[0].sender()?, DEFAULT_TX_VALUE)
+        .await?;
+    provider
+        .assert_balance(receipts[1].sender()?, DEFAULT_TX_VALUE)
+        .await?;
+
+    // Assert we can still finalize transactions after loading state
+    provider.tx().finalize().await?;
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn cant_load_into_existing_state() -> anyhow::Result<()> {
+    // Test that we can't load new state into a node with existing state.
+    let provider = init_testing_provider(identity).await?;
+
+    let old_receipts = [
+        provider.tx().finalize().await?,
+        provider.tx().finalize().await?,
+    ];
+    let old_blocks = provider.get_blocks_by_receipts(&old_receipts).await?;
+
+    // Dump node's state and re-create it
+    let state = provider.anvil_dump_state().await?;
+    let provider = init_testing_provider(identity).await?;
+
+    let new_receipts = [
+        provider.tx().finalize().await?,
+        provider.tx().finalize().await?,
+    ];
+    let new_blocks = provider.get_blocks_by_receipts(&new_receipts).await?;
+
+    // Load state into the new node, make sure it fails and assert that the node still has new
+    // receipts, blocks and state.
+    assert!(provider.anvil_load_state(state).await.is_err());
+    provider.assert_has_receipts(&new_receipts).await?;
+    provider.assert_has_blocks(&new_blocks).await?;
+    provider
+        .assert_balance(new_receipts[0].sender()?, DEFAULT_TX_VALUE)
+        .await?;
+    provider
+        .assert_balance(new_receipts[1].sender()?, DEFAULT_TX_VALUE)
+        .await?;
+
+    // Assert the node does not have old state
+    provider.assert_no_receipts(&old_receipts).await?;
+    provider.assert_no_blocks(&old_blocks).await?;
+    provider
+        .assert_balance(old_receipts[0].sender()?, 0)
+        .await?;
+    provider
+        .assert_balance(old_receipts[1].sender()?, 0)
+        .await?;
+
+    Ok(())
+}
diff --git a/src/testing.rs b/src/testing.rs
index 1ed388d..55aeec3 100644
--- a/src/testing.rs
+++ b/src/testing.rs
@@ -19,7 +19,6 @@ use httptest::{
 };
 use itertools::Itertools;
 use std::str::FromStr;
-use zksync_multivm::interface::{ExecutionResult, VmExecutionResultAndLogs};
 use zksync_types::api::{
     BlockDetailsBase, BlockIdVariant, BlockStatus, BridgeAddresses, DebugCall, DebugCallType, Log,
 };
@@ -669,13 +668,6 @@ pub fn default_tx_execution_info() -> TxExecutionInfo {
         },
         batch_number: Default::default(),
         miniblock_number: Default::default(),
-        result: VmExecutionResultAndLogs {
-            result: ExecutionResult::Success { output: vec![] },
-            logs: Default::default(),
-            statistics: Default::default(),
-            refunds: Default::default(),
-            new_known_factory_deps: None,
-        },
     }
 }