diff --git a/Cargo.lock b/Cargo.lock
index fcef5af1ca68..f6c07b39dc67 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -9746,6 +9746,7 @@ dependencies = [
  "once_cell",
  "rand 0.8.5",
  "tempfile",
+ "test-casing",
  "tokio",
  "tracing",
  "vise",
diff --git a/core/bin/external_node/src/main.rs b/core/bin/external_node/src/main.rs
index f6696d733482..f412b6006cc2 100644
--- a/core/bin/external_node/src/main.rs
+++ b/core/bin/external_node/src/main.rs
@@ -94,8 +94,8 @@ async fn build_state_keeper(
         stop_receiver_clone.changed().await?;
         result
     }));
-    let batch_executor_base: Box<dyn BatchExecutor> =
-        Box::new(MainBatchExecutor::new(save_call_traces, true));
+    let batch_executor = MainBatchExecutor::new(save_call_traces, true);
+    let batch_executor: Box<dyn BatchExecutor> = Box::new(batch_executor);
 
     let io = ExternalIO::new(
         connection_pool,
@@ -108,7 +108,7 @@ async fn build_state_keeper(
     Ok(ZkSyncStateKeeper::new(
         stop_receiver,
         Box::new(io),
-        batch_executor_base,
+        batch_executor,
         output_handler,
         Arc::new(NoopSealer),
         Arc::new(storage_factory),
diff --git a/core/bin/zksync_server/src/main.rs b/core/bin/zksync_server/src/main.rs
index a59705b8e587..ee29f48c486a 100644
--- a/core/bin/zksync_server/src/main.rs
+++ b/core/bin/zksync_server/src/main.rs
@@ -12,10 +12,10 @@ use zksync_config::{
         fri_prover_group::FriProverGroupConfig,
         house_keeper::HouseKeeperConfig,
         BasicWitnessInputProducerConfig, ContractsConfig, DatabaseSecrets,
-        ExternalPriceApiClientConfig, FriProofCompressorConfig, FriProverConfig,
-        FriProverGatewayConfig, FriWitnessGeneratorConfig, FriWitnessVectorGeneratorConfig,
-        L1Secrets, ObservabilityConfig, PrometheusConfig, ProofDataHandlerConfig,
-        ProtectiveReadsWriterConfig, Secrets,
+        ExperimentalVmPlaygroundConfig, ExternalPriceApiClientConfig, FriProofCompressorConfig,
+        FriProverConfig, FriProverGatewayConfig, FriWitnessGeneratorConfig,
+        FriWitnessVectorGeneratorConfig, L1Secrets, ObservabilityConfig, PrometheusConfig,
+        ProofDataHandlerConfig, ProtectiveReadsWriterConfig, Secrets,
     },
     ApiConfig, BaseTokenAdjusterConfig, ContractVerifierConfig, DADispatcherConfig, DBConfig,
     EthConfig, EthWatchConfig, GasAdjusterConfig, GenesisConfig, ObjectStoreConfig, PostgresConfig,
@@ -230,6 +230,7 @@ fn load_env_config() -> anyhow::Result<TempConfigStore> {
         da_dispatcher_config: DADispatcherConfig::from_env().ok(),
         protective_reads_writer_config: ProtectiveReadsWriterConfig::from_env().ok(),
         basic_witness_input_producer_config: BasicWitnessInputProducerConfig::from_env().ok(),
+        vm_playground_config: ExperimentalVmPlaygroundConfig::from_env().ok(),
         core_object_store: ObjectStoreConfig::from_env().ok(),
         base_token_adjuster_config: BaseTokenAdjusterConfig::from_env().ok(),
         commitment_generator: None,
diff --git a/core/bin/zksync_server/src/node_builder.rs b/core/bin/zksync_server/src/node_builder.rs
index 64039ddcc873..c15fb13c4665 100644
--- a/core/bin/zksync_server/src/node_builder.rs
+++ b/core/bin/zksync_server/src/node_builder.rs
@@ -53,7 +53,8 @@ use zksync_node_framework::{
         },
         tee_verifier_input_producer::TeeVerifierInputProducerLayer,
         vm_runner::{
-            bwip::BasicWitnessInputProducerLayer, protective_reads::ProtectiveReadsWriterLayer,
+            bwip::BasicWitnessInputProducerLayer, playground::VmPlaygroundLayer,
+            protective_reads::ProtectiveReadsWriterLayer,
         },
         web3_api::{
             caches::MempoolCacheLayer,
@@ -550,6 +551,16 @@ impl MainNodeBuilder {
         Ok(self)
     }
 
+    fn add_vm_playground_layer(mut self) -> anyhow::Result<Self> {
+        let vm_playground_config = try_load_config!(self.configs.vm_playground_config);
+        self.node.add_layer(VmPlaygroundLayer::new(
+            vm_playground_config,
+            self.genesis_config.l2_chain_id,
+        ));
+
+        Ok(self)
+    }
+
     fn add_base_token_ratio_persister_layer(mut self) -> anyhow::Result<Self> {
         let config = try_load_config!(self.configs.base_token_adjuster);
         let contracts_config = self.contracts_config.clone();
@@ -700,6 +711,9 @@ impl MainNodeBuilder {
                 Component::VmRunnerBwip => {
                     self = self.add_vm_runner_bwip_layer()?;
                 }
+                Component::VmPlayground => {
+                    self = self.add_vm_playground_layer()?;
+                }
             }
         }
         Ok(self.node.build()?)
diff --git a/core/lib/basic_types/src/lib.rs b/core/lib/basic_types/src/lib.rs
index 21e90f4bad77..be295a003fd2 100644
--- a/core/lib/basic_types/src/lib.rs
+++ b/core/lib/basic_types/src/lib.rs
@@ -28,7 +28,7 @@ pub mod protocol_version;
 pub mod prover_dal;
 pub mod tee_types;
 pub mod url;
-pub mod vm_version;
+pub mod vm;
 pub mod web3;
 
 /// Account place in the global state tree is uniquely identified by its address.
diff --git a/core/lib/basic_types/src/protocol_version.rs b/core/lib/basic_types/src/protocol_version.rs
index f0d12436e3b8..e95fdfb57eb5 100644
--- a/core/lib/basic_types/src/protocol_version.rs
+++ b/core/lib/basic_types/src/protocol_version.rs
@@ -12,7 +12,7 @@ use serde_with::{DeserializeFromStr, SerializeDisplay};
 
 use crate::{
     ethabi::Token,
-    vm_version::VmVersion,
+    vm::VmVersion,
     web3::contract::{Detokenize, Error},
     H256, U256,
 };
diff --git a/core/lib/basic_types/src/vm.rs b/core/lib/basic_types/src/vm.rs
new file mode 100644
index 000000000000..c178c853b2dc
--- /dev/null
+++ b/core/lib/basic_types/src/vm.rs
@@ -0,0 +1,39 @@
+//! Basic VM types that shared widely enough to not put them in the `multivm` crate.
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Copy)]
+pub enum VmVersion {
+    M5WithoutRefunds,
+    M5WithRefunds,
+    M6Initial,
+    M6BugWithCompressionFixed,
+    Vm1_3_2,
+    VmVirtualBlocks,
+    VmVirtualBlocksRefundsEnhancement,
+    VmBoojumIntegration,
+    Vm1_4_1,
+    Vm1_4_2,
+    Vm1_5_0SmallBootloaderMemory,
+    Vm1_5_0IncreasedBootloaderMemory,
+}
+
+impl VmVersion {
+    /// Returns the latest supported VM version.
+    pub const fn latest() -> VmVersion {
+        Self::Vm1_5_0IncreasedBootloaderMemory
+    }
+}
+
+/// Mode in which to run the new fast VM implementation.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum FastVmMode {
+    /// Run only the old VM.
+    #[default]
+    Old,
+    /// Run only the new Vm.
+    New,
+    /// Run both the new and old VM and compare their outputs for each transaction execution.
+    Shadow,
+}
diff --git a/core/lib/basic_types/src/vm_version.rs b/core/lib/basic_types/src/vm_version.rs
deleted file mode 100644
index 49fec39fc9cb..000000000000
--- a/core/lib/basic_types/src/vm_version.rs
+++ /dev/null
@@ -1,22 +0,0 @@
-#[derive(Debug, Clone, Copy)]
-pub enum VmVersion {
-    M5WithoutRefunds,
-    M5WithRefunds,
-    M6Initial,
-    M6BugWithCompressionFixed,
-    Vm1_3_2,
-    VmVirtualBlocks,
-    VmVirtualBlocksRefundsEnhancement,
-    VmBoojumIntegration,
-    Vm1_4_1,
-    Vm1_4_2,
-    Vm1_5_0SmallBootloaderMemory,
-    Vm1_5_0IncreasedBootloaderMemory,
-}
-
-impl VmVersion {
-    /// Returns the latest supported VM version.
-    pub const fn latest() -> VmVersion {
-        Self::Vm1_5_0IncreasedBootloaderMemory
-    }
-}
diff --git a/core/lib/config/src/configs/experimental.rs b/core/lib/config/src/configs/experimental.rs
index e362715d3d4a..0b9505d4e19b 100644
--- a/core/lib/config/src/configs/experimental.rs
+++ b/core/lib/config/src/configs/experimental.rs
@@ -3,6 +3,7 @@
 use std::num::NonZeroU32;
 
 use serde::Deserialize;
+use zksync_basic_types::{vm::FastVmMode, L1BatchNumber};
 
 #[derive(Debug, Clone, PartialEq, Deserialize)]
 pub struct ExperimentalDBConfig {
@@ -60,3 +61,37 @@ impl ExperimentalDBConfig {
         100
     }
 }
+
+#[derive(Debug, Deserialize, Clone, PartialEq)]
+pub struct ExperimentalVmPlaygroundConfig {
+    /// Mode in which to run the fast VM implementation. Note that for it to actually be used, L1 batches should have a recent version.
+    #[serde(default)]
+    pub fast_vm_mode: FastVmMode,
+    /// Path to the RocksDB cache directory.
+    #[serde(default = "ExperimentalVmPlaygroundConfig::default_db_path")]
+    pub db_path: String,
+    /// First L1 batch to consider processed. Will not be used if the processing cursor is persisted, unless the `reset` flag is set.
+    #[serde(default)]
+    pub first_processed_batch: L1BatchNumber,
+    /// If set to true, processing cursor will reset `first_processed_batch` regardless of the current progress. Beware that this will likely
+    /// require to drop the RocksDB cache.
+    #[serde(default)]
+    pub reset: bool,
+}
+
+impl Default for ExperimentalVmPlaygroundConfig {
+    fn default() -> Self {
+        Self {
+            fast_vm_mode: FastVmMode::default(),
+            db_path: Self::default_db_path(),
+            first_processed_batch: L1BatchNumber(0),
+            reset: false,
+        }
+    }
+}
+
+impl ExperimentalVmPlaygroundConfig {
+    pub fn default_db_path() -> String {
+        "./db/vm_playground".to_owned()
+    }
+}
diff --git a/core/lib/config/src/configs/general.rs b/core/lib/config/src/configs/general.rs
index 122d1e278553..133391b34975 100644
--- a/core/lib/config/src/configs/general.rs
+++ b/core/lib/config/src/configs/general.rs
@@ -9,10 +9,10 @@ use crate::{
         pruning::PruningConfig,
         snapshot_recovery::SnapshotRecoveryConfig,
         vm_runner::{BasicWitnessInputProducerConfig, ProtectiveReadsWriterConfig},
-        CommitmentGeneratorConfig, ExternalPriceApiClientConfig, FriProofCompressorConfig,
-        FriProverConfig, FriProverGatewayConfig, FriWitnessGeneratorConfig,
-        FriWitnessVectorGeneratorConfig, ObservabilityConfig, PrometheusConfig,
-        ProofDataHandlerConfig,
+        CommitmentGeneratorConfig, ExperimentalVmPlaygroundConfig, ExternalPriceApiClientConfig,
+        FriProofCompressorConfig, FriProverConfig, FriProverGatewayConfig,
+        FriWitnessGeneratorConfig, FriWitnessVectorGeneratorConfig, ObservabilityConfig,
+        PrometheusConfig, ProofDataHandlerConfig,
     },
     ApiConfig, ContractVerifierConfig, DBConfig, EthConfig, ObjectStoreConfig, PostgresConfig,
     SnapshotsCreatorConfig,
@@ -43,6 +43,7 @@ pub struct GeneralConfig {
     pub da_dispatcher_config: Option<DADispatcherConfig>,
     pub protective_reads_writer_config: Option<ProtectiveReadsWriterConfig>,
     pub basic_witness_input_producer_config: Option<BasicWitnessInputProducerConfig>,
+    pub vm_playground_config: Option<ExperimentalVmPlaygroundConfig>,
     pub commitment_generator: Option<CommitmentGeneratorConfig>,
     pub snapshot_recovery: Option<SnapshotRecoveryConfig>,
     pub pruning: Option<PruningConfig>,
diff --git a/core/lib/config/src/configs/mod.rs b/core/lib/config/src/configs/mod.rs
index 0da6f986f353..06c5a8af97f0 100644
--- a/core/lib/config/src/configs/mod.rs
+++ b/core/lib/config/src/configs/mod.rs
@@ -9,7 +9,7 @@ pub use self::{
     database::{DBConfig, PostgresConfig},
     eth_sender::{EthConfig, GasAdjusterConfig},
     eth_watch::EthWatchConfig,
-    experimental::ExperimentalDBConfig,
+    experimental::{ExperimentalDBConfig, ExperimentalVmPlaygroundConfig},
     external_price_api_client::ExternalPriceApiClientConfig,
     fri_proof_compressor::FriProofCompressorConfig,
     fri_prover::FriProverConfig,
diff --git a/core/lib/config/src/configs/vm_runner.rs b/core/lib/config/src/configs/vm_runner.rs
index fa7c7c1a90a3..1fecc12668c1 100644
--- a/core/lib/config/src/configs/vm_runner.rs
+++ b/core/lib/config/src/configs/vm_runner.rs
@@ -1,7 +1,7 @@
 use serde::Deserialize;
 use zksync_basic_types::L1BatchNumber;
 
-#[derive(Debug, Deserialize, Clone, PartialEq, Default)]
+#[derive(Debug, Deserialize, Clone, PartialEq)]
 pub struct ProtectiveReadsWriterConfig {
     /// Path to the RocksDB data directory that serves state cache.
     #[serde(default = "ProtectiveReadsWriterConfig::default_db_path")]
@@ -18,7 +18,7 @@ impl ProtectiveReadsWriterConfig {
     }
 }
 
-#[derive(Debug, Deserialize, Clone, PartialEq, Default)]
+#[derive(Debug, Deserialize, Clone, PartialEq)]
 pub struct BasicWitnessInputProducerConfig {
     /// Path to the RocksDB data directory that serves state cache.
     #[serde(default = "BasicWitnessInputProducerConfig::default_db_path")]
diff --git a/core/lib/config/src/testonly.rs b/core/lib/config/src/testonly.rs
index f3d6b98491be..42ce4ed63891 100644
--- a/core/lib/config/src/testonly.rs
+++ b/core/lib/config/src/testonly.rs
@@ -6,6 +6,7 @@ use zksync_basic_types::{
     commitment::L1BatchCommitmentMode,
     network::Network,
     protocol_version::{ProtocolSemanticVersion, ProtocolVersionId, VersionPatch},
+    vm::FastVmMode,
     L1BatchNumber, L1ChainId, L2ChainId,
 };
 use zksync_consensus_utils::EncodeDist;
@@ -293,6 +294,21 @@ impl Distribution<configs::ExperimentalDBConfig> for EncodeDist {
     }
 }
 
+impl Distribution<configs::ExperimentalVmPlaygroundConfig> for EncodeDist {
+    fn sample<R: Rng + ?Sized>(&self, rng: &mut R) -> configs::ExperimentalVmPlaygroundConfig {
+        configs::ExperimentalVmPlaygroundConfig {
+            fast_vm_mode: match rng.gen_range(0..3) {
+                0 => FastVmMode::Old,
+                1 => FastVmMode::New,
+                _ => FastVmMode::Shadow,
+            },
+            db_path: self.sample(rng),
+            first_processed_batch: L1BatchNumber(rng.gen()),
+            reset: self.sample(rng),
+        }
+    }
+}
+
 impl Distribution<configs::database::DBConfig> for EncodeDist {
     fn sample<R: Rng + ?Sized>(&self, rng: &mut R) -> configs::database::DBConfig {
         configs::database::DBConfig {
@@ -1038,6 +1054,7 @@ impl Distribution<configs::GeneralConfig> for EncodeDist {
             da_dispatcher_config: self.sample(rng),
             protective_reads_writer_config: self.sample(rng),
             basic_witness_input_producer_config: self.sample(rng),
+            vm_playground_config: self.sample(rng),
             commitment_generator: self.sample(rng),
             snapshot_recovery: self.sample(rng),
             pruning: self.sample(rng),
diff --git a/core/lib/env_config/src/vm_runner.rs b/core/lib/env_config/src/vm_runner.rs
index 9973d760a236..49ed5d8b9289 100644
--- a/core/lib/env_config/src/vm_runner.rs
+++ b/core/lib/env_config/src/vm_runner.rs
@@ -1,4 +1,6 @@
-use zksync_config::configs::{BasicWitnessInputProducerConfig, ProtectiveReadsWriterConfig};
+use zksync_config::configs::{
+    BasicWitnessInputProducerConfig, ExperimentalVmPlaygroundConfig, ProtectiveReadsWriterConfig,
+};
 
 use crate::{envy_load, FromEnv};
 
@@ -13,3 +15,69 @@ impl FromEnv for BasicWitnessInputProducerConfig {
         envy_load("vm_runner.bwip", "VM_RUNNER_BWIP_")
     }
 }
+
+impl FromEnv for ExperimentalVmPlaygroundConfig {
+    fn from_env() -> anyhow::Result<Self> {
+        envy_load("vm_runner.playground", "VM_RUNNER_PLAYGROUND_")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use zksync_basic_types::{vm::FastVmMode, L1BatchNumber};
+
+    use super::*;
+    use crate::test_utils::EnvMutex;
+
+    static MUTEX: EnvMutex = EnvMutex::new();
+
+    #[test]
+    fn bwip_config_from_env() {
+        let mut lock = MUTEX.lock();
+        let config = r#"
+            VM_RUNNER_BWIP_DB_PATH=/db/bwip
+            VM_RUNNER_BWIP_WINDOW_SIZE=50
+            VM_RUNNER_BWIP_FIRST_PROCESSED_BATCH=123
+        "#;
+        lock.set_env(config);
+
+        let config = BasicWitnessInputProducerConfig::from_env().unwrap();
+        assert_eq!(config.db_path, "/db/bwip");
+        assert_eq!(config.window_size, 50);
+        assert_eq!(config.first_processed_batch, L1BatchNumber(123));
+    }
+
+    #[test]
+    fn playground_config_from_env() {
+        let mut lock = MUTEX.lock();
+        let config = r#"
+            VM_RUNNER_PLAYGROUND_FAST_VM_MODE=shadow
+            VM_RUNNER_PLAYGROUND_DB_PATH=/db/vm_playground
+            VM_RUNNER_PLAYGROUND_FIRST_PROCESSED_BATCH=123
+            VM_RUNNER_PLAYGROUND_RESET=true
+        "#;
+        lock.set_env(config);
+
+        let config = ExperimentalVmPlaygroundConfig::from_env().unwrap();
+        assert_eq!(config.fast_vm_mode, FastVmMode::Shadow);
+        assert_eq!(config.db_path, "/db/vm_playground");
+        assert_eq!(config.first_processed_batch, L1BatchNumber(123));
+        assert!(config.reset);
+
+        lock.remove_env(&["VM_RUNNER_PLAYGROUND_RESET"]);
+        let config = ExperimentalVmPlaygroundConfig::from_env().unwrap();
+        assert!(!config.reset);
+
+        lock.remove_env(&["VM_RUNNER_PLAYGROUND_FIRST_PROCESSED_BATCH"]);
+        let config = ExperimentalVmPlaygroundConfig::from_env().unwrap();
+        assert_eq!(config.first_processed_batch, L1BatchNumber(0));
+
+        lock.remove_env(&["VM_RUNNER_PLAYGROUND_FAST_VM_MODE"]);
+        let config = ExperimentalVmPlaygroundConfig::from_env().unwrap();
+        assert_eq!(config.fast_vm_mode, FastVmMode::Old);
+
+        lock.remove_env(&["VM_RUNNER_PLAYGROUND_DB_PATH"]);
+        let config = ExperimentalVmPlaygroundConfig::from_env().unwrap();
+        assert!(!config.db_path.is_empty());
+    }
+}
diff --git a/core/lib/multivm/src/lib.rs b/core/lib/multivm/src/lib.rs
index 92b6c759d968..08b077ce3eab 100644
--- a/core/lib/multivm/src/lib.rs
+++ b/core/lib/multivm/src/lib.rs
@@ -4,17 +4,17 @@
 
 pub use circuit_sequencer_api_1_5_0 as circuit_sequencer_api_latest;
 pub use zk_evm_1_5_0 as zk_evm_latest;
-pub use zksync_types::vm_version::VmVersion;
+pub use zksync_types::vm::VmVersion;
 
-pub use self::versions::{
-    vm_1_3_2, vm_1_4_1, vm_1_4_2, vm_boojum_integration, vm_fast, vm_latest, vm_m5, vm_m6,
-    vm_refunds_enhancement, vm_virtual_blocks,
-};
 pub use crate::{
     glue::{
         history_mode::HistoryMode,
         tracers::{MultiVMTracer, MultiVmTracerPointer},
     },
+    versions::{
+        vm_1_3_2, vm_1_4_1, vm_1_4_2, vm_boojum_integration, vm_fast, vm_latest, vm_m5, vm_m6,
+        vm_refunds_enhancement, vm_virtual_blocks,
+    },
     vm_instance::VmInstance,
 };
 
diff --git a/core/lib/multivm/src/tracers/validator/mod.rs b/core/lib/multivm/src/tracers/validator/mod.rs
index b56d92015a33..635915f95278 100644
--- a/core/lib/multivm/src/tracers/validator/mod.rs
+++ b/core/lib/multivm/src/tracers/validator/mod.rs
@@ -7,8 +7,8 @@ use zksync_system_constants::{
     L2_BASE_TOKEN_ADDRESS, MSG_VALUE_SIMULATOR_ADDRESS, SYSTEM_CONTEXT_ADDRESS,
 };
 use zksync_types::{
-    vm_trace::ViolatedValidationRule, web3::keccak256, AccountTreeId, Address, StorageKey,
-    VmVersion, H256, U256,
+    vm::VmVersion, vm_trace::ViolatedValidationRule, web3::keccak256, AccountTreeId, Address,
+    StorageKey, H256, U256,
 };
 use zksync_utils::{be_bytes_to_safe_address, u256_to_account_address, u256_to_h256};
 
diff --git a/core/lib/multivm/src/utils.rs b/core/lib/multivm/src/utils.rs
index a15fdba6b703..96ae580a5f73 100644
--- a/core/lib/multivm/src/utils.rs
+++ b/core/lib/multivm/src/utils.rs
@@ -1,6 +1,7 @@
 use zksync_types::{
     fee_model::{BatchFeeInput, L1PeggedBatchFeeModelInput, PubdataIndependentBatchFeeModelInput},
-    VmVersion, U256,
+    vm::VmVersion,
+    U256,
 };
 
 use crate::vm_latest::L1BatchEnv;
diff --git a/core/lib/multivm/src/versions/vm_fast/mod.rs b/core/lib/multivm/src/versions/vm_fast/mod.rs
index 4b4228afb2f3..4deb6b9dbf74 100644
--- a/core/lib/multivm/src/versions/vm_fast/mod.rs
+++ b/core/lib/multivm/src/versions/vm_fast/mod.rs
@@ -1,3 +1,5 @@
+pub use self::vm::Vm;
+
 mod bootloader_state;
 mod bytecode;
 mod events;
@@ -10,5 +12,3 @@ mod refund;
 mod tests;
 mod transaction_data;
 mod vm;
-
-pub use vm::Vm;
diff --git a/core/lib/multivm/src/versions/vm_latest/vm.rs b/core/lib/multivm/src/versions/vm_latest/vm.rs
index 6d7ce8b32bab..f11431f01546 100644
--- a/core/lib/multivm/src/versions/vm_latest/vm.rs
+++ b/core/lib/multivm/src/versions/vm_latest/vm.rs
@@ -3,7 +3,8 @@ use zksync_state::{StoragePtr, WriteStorage};
 use zksync_types::{
     event::extract_l2tol1logs_from_l1_messenger,
     l2_to_l1_log::{SystemL2ToL1Log, UserL2ToL1Log},
-    Transaction, VmVersion,
+    vm::VmVersion,
+    Transaction,
 };
 use zksync_utils::bytecode::CompressedBytecodeInfo;
 
diff --git a/core/lib/multivm/src/versions/vm_m5/vm.rs b/core/lib/multivm/src/versions/vm_m5/vm.rs
index 223d875d7906..53189dbcfef5 100644
--- a/core/lib/multivm/src/versions/vm_m5/vm.rs
+++ b/core/lib/multivm/src/versions/vm_m5/vm.rs
@@ -4,7 +4,8 @@ use zk_evm_1_3_1::aux_structures::LogQuery;
 use zksync_state::StoragePtr;
 use zksync_types::{
     l2_to_l1_log::{L2ToL1Log, UserL2ToL1Log},
-    Transaction, VmVersion,
+    vm::VmVersion,
+    Transaction,
 };
 use zksync_utils::{bytecode::CompressedBytecodeInfo, h256_to_u256, u256_to_h256};
 
diff --git a/core/lib/multivm/src/versions/vm_m6/vm.rs b/core/lib/multivm/src/versions/vm_m6/vm.rs
index 6eee706151f3..634867697a92 100644
--- a/core/lib/multivm/src/versions/vm_m6/vm.rs
+++ b/core/lib/multivm/src/versions/vm_m6/vm.rs
@@ -6,7 +6,8 @@ use zk_evm_1_3_1::aux_structures::LogQuery;
 use zksync_state::StoragePtr;
 use zksync_types::{
     l2_to_l1_log::{L2ToL1Log, UserL2ToL1Log},
-    Transaction, VmVersion,
+    vm::VmVersion,
+    Transaction,
 };
 use zksync_utils::{
     bytecode::{hash_bytecode, CompressedBytecodeInfo},
diff --git a/core/lib/multivm/src/vm_instance.rs b/core/lib/multivm/src/vm_instance.rs
index 67e9a8d52540..c8a7ce837991 100644
--- a/core/lib/multivm/src/vm_instance.rs
+++ b/core/lib/multivm/src/vm_instance.rs
@@ -1,5 +1,5 @@
-use zksync_state::{ReadStorage, StoragePtr, StorageView};
-use zksync_types::VmVersion;
+use zksync_state::{ImmutableStorageView, ReadStorage, StoragePtr, StorageView};
+use zksync_types::vm::{FastVmMode, VmVersion};
 use zksync_utils::bytecode::CompressedBytecodeInfo;
 
 use crate::{
@@ -13,7 +13,7 @@ use crate::{
     versions::shadow::ShadowVm,
 };
 
-pub type FastVm<S, H> = ShadowVm<S, crate::vm_latest::Vm<StorageView<S>, H>>;
+pub type ShadowedFastVm<S, H> = ShadowVm<S, crate::vm_latest::Vm<StorageView<S>, H>>;
 
 #[derive(Debug)]
 pub enum VmInstance<S: ReadStorage, H: HistoryMode> {
@@ -26,7 +26,8 @@ pub enum VmInstance<S: ReadStorage, H: HistoryMode> {
     Vm1_4_1(crate::vm_1_4_1::Vm<StorageView<S>, H>),
     Vm1_4_2(crate::vm_1_4_2::Vm<StorageView<S>, H>),
     Vm1_5_0(crate::vm_latest::Vm<StorageView<S>, H>),
-    VmFast(FastVm<S, H>),
+    VmFast(crate::vm_fast::Vm<ImmutableStorageView<S>>),
+    ShadowedVmFast(ShadowedFastVm<S, H>),
 }
 
 macro_rules! dispatch_vm {
@@ -42,6 +43,7 @@ macro_rules! dispatch_vm {
             VmInstance::Vm1_4_2(vm) => vm.$function($($params)*),
             VmInstance::Vm1_5_0(vm) => vm.$function($($params)*),
             VmInstance::VmFast(vm) => vm.$function($($params)*),
+            VmInstance::ShadowedVmFast(vm) => vm.$function($($params)*),
         }
     };
 }
@@ -137,8 +139,7 @@ impl<S: ReadStorage, H: HistoryMode> VmFactory<StorageView<S>> for VmInstance<S,
     ) -> Self {
         let protocol_version = system_env.version;
         let vm_version: VmVersion = protocol_version.into();
-        //Self::new_with_specific_version(batch_env, system_env, storage_view, vm_version)
-        Self::new_fast_with_specific_version(batch_env, system_env, storage_view, vm_version)
+        Self::new_with_specific_version(batch_env, system_env, storage_view, vm_version)
     }
 }
 
@@ -247,18 +248,26 @@ impl<S: ReadStorage, H: HistoryMode> VmInstance<S, H> {
         }
     }
 
-    fn new_fast_with_specific_version(
+    /// Creates a VM that may use the fast VM depending on the protocol version in `system_env` and `mode`.
+    pub fn maybe_fast(
         l1_batch_env: L1BatchEnv,
         system_env: SystemEnv,
         storage_view: StoragePtr<StorageView<S>>,
-        vm_version: VmVersion,
+        mode: FastVmMode,
     ) -> Self {
+        let vm_version = system_env.version.into();
         match vm_version {
-            VmVersion::Vm1_5_0IncreasedBootloaderMemory => VmInstance::VmFast(
-                //crate::vm_fast::Vm::new(l1_batch_env, system_env, storage_view),
-                FastVm::new(l1_batch_env, system_env, storage_view),
-            ),
-            _ => unimplemented!("version not supported by fast VM"),
+            VmVersion::Vm1_5_0IncreasedBootloaderMemory => match mode {
+                FastVmMode::Old => Self::new(l1_batch_env, system_env, storage_view),
+                FastVmMode::New => {
+                    let storage = ImmutableStorageView::new(storage_view);
+                    Self::VmFast(crate::vm_fast::Vm::new(l1_batch_env, system_env, storage))
+                }
+                FastVmMode::Shadow => {
+                    Self::ShadowedVmFast(ShadowVm::new(l1_batch_env, system_env, storage_view))
+                }
+            },
+            _ => Self::new(l1_batch_env, system_env, storage_view),
         }
     }
 }
diff --git a/core/lib/protobuf_config/src/experimental.rs b/core/lib/protobuf_config/src/experimental.rs
index 8d92f3ef87a8..fa49e4836456 100644
--- a/core/lib/protobuf_config/src/experimental.rs
+++ b/core/lib/protobuf_config/src/experimental.rs
@@ -1,6 +1,7 @@
 use std::num::NonZeroU32;
 
 use anyhow::Context as _;
+use zksync_basic_types::{vm::FastVmMode, L1BatchNumber};
 use zksync_config::configs;
 use zksync_protobuf::{repr::ProtoRepr, required};
 
@@ -49,3 +50,51 @@ impl ProtoRepr for proto::Db {
         }
     }
 }
+
+impl proto::FastVmMode {
+    fn new(source: FastVmMode) -> Self {
+        match source {
+            FastVmMode::Old => Self::Old,
+            FastVmMode::New => Self::New,
+            FastVmMode::Shadow => Self::Shadow,
+        }
+    }
+
+    fn parse(&self) -> FastVmMode {
+        match self {
+            Self::Old => FastVmMode::Old,
+            Self::New => FastVmMode::New,
+            Self::Shadow => FastVmMode::Shadow,
+        }
+    }
+}
+
+impl ProtoRepr for proto::VmPlayground {
+    type Type = configs::ExperimentalVmPlaygroundConfig;
+
+    fn read(&self) -> anyhow::Result<Self::Type> {
+        Ok(Self::Type {
+            fast_vm_mode: self
+                .fast_vm_mode
+                .map(proto::FastVmMode::try_from)
+                .transpose()
+                .context("fast_vm_mode")?
+                .map_or_else(FastVmMode::default, |mode| mode.parse()),
+            db_path: self
+                .db_path
+                .clone()
+                .unwrap_or_else(Self::Type::default_db_path),
+            first_processed_batch: L1BatchNumber(self.first_processed_batch.unwrap_or(0)),
+            reset: self.reset.unwrap_or(false),
+        })
+    }
+
+    fn build(this: &Self::Type) -> Self {
+        Self {
+            fast_vm_mode: Some(proto::FastVmMode::new(this.fast_vm_mode).into()),
+            db_path: Some(this.db_path.clone()),
+            first_processed_batch: Some(this.first_processed_batch.0),
+            reset: Some(this.reset),
+        }
+    }
+}
diff --git a/core/lib/protobuf_config/src/general.rs b/core/lib/protobuf_config/src/general.rs
index 31d1ea6bc1b7..52b39542fee3 100644
--- a/core/lib/protobuf_config/src/general.rs
+++ b/core/lib/protobuf_config/src/general.rs
@@ -57,6 +57,8 @@ impl ProtoRepr for proto::GeneralConfig {
             external_price_api_client_config: read_optional_repr(&self.external_price_api_client)
                 .context("external_price_api_client")?,
             consensus_config: read_optional_repr(&self.consensus).context("consensus")?,
+            vm_playground_config: read_optional_repr(&self.vm_playground)
+                .context("vm_playground")?,
         })
     }
 
@@ -107,6 +109,7 @@ impl ProtoRepr for proto::GeneralConfig {
                 .as_ref()
                 .map(ProtoRepr::build),
             consensus: this.consensus_config.as_ref().map(ProtoRepr::build),
+            vm_playground: this.vm_playground_config.as_ref().map(ProtoRepr::build),
         }
     }
 }
diff --git a/core/lib/protobuf_config/src/proto/config/experimental.proto b/core/lib/protobuf_config/src/proto/config/experimental.proto
index 1336c4719d26..fb9342a166ad 100644
--- a/core/lib/protobuf_config/src/proto/config/experimental.proto
+++ b/core/lib/protobuf_config/src/proto/config/experimental.proto
@@ -18,3 +18,17 @@ message SnapshotRecovery {
   optional uint64 tree_recovery_parallel_persistence_buffer = 1;
   optional bool drop_storage_key_preimages = 2; // optional; false by default
 }
+
+enum FastVmMode {
+  OLD = 0;
+  NEW = 1;
+  SHADOW = 2;
+}
+
+// Experimental VM configuration
+message VmPlayground {
+  optional FastVmMode fast_vm_mode = 1; // optional; if not set, fast VM is not used
+  optional string db_path = 2; // optional; defaults to `./db/vm_playground`
+  optional uint32 first_processed_batch = 3; // optional; defaults to 0
+  optional bool reset = 4; // optional; defaults to false
+}
diff --git a/core/lib/protobuf_config/src/proto/config/general.proto b/core/lib/protobuf_config/src/proto/config/general.proto
index 37d507b9ab62..80590c3df0a8 100644
--- a/core/lib/protobuf_config/src/proto/config/general.proto
+++ b/core/lib/protobuf_config/src/proto/config/general.proto
@@ -9,6 +9,7 @@ import "zksync/config/contract_verifier.proto";
 import "zksync/config/database.proto";
 import "zksync/config/circuit_breaker.proto";
 import "zksync/config/eth_sender.proto";
+import "zksync/config/experimental.proto";
 import "zksync/config/house_keeper.proto";
 import "zksync/config/observability.proto";
 import "zksync/config/snapshots_creator.proto";
@@ -54,4 +55,5 @@ message GeneralConfig {
   optional vm_runner.BasicWitnessInputProducer basic_witness_input_producer = 40;
   optional external_price_api_client.ExternalPriceApiClient external_price_api_client = 41;
   optional core.consensus.Config consensus = 42;
+  optional experimental.VmPlayground vm_playground = 43;
 }
diff --git a/core/lib/protobuf_config/src/proto/config/vm_runner.proto b/core/lib/protobuf_config/src/proto/config/vm_runner.proto
index 93521a5fd893..d6537c109e6d 100644
--- a/core/lib/protobuf_config/src/proto/config/vm_runner.proto
+++ b/core/lib/protobuf_config/src/proto/config/vm_runner.proto
@@ -2,6 +2,8 @@ syntax = "proto3";
 
 package zksync.config.vm_runner;
 
+import "zksync/config/experimental.proto";
+
 message ProtectiveReadsWriter {
   optional string db_path = 1; // required; fs path
   optional uint64 window_size = 2; // required
diff --git a/core/lib/protobuf_config/src/vm_runner.rs b/core/lib/protobuf_config/src/vm_runner.rs
index cc0d53ad519e..134cc20952f1 100644
--- a/core/lib/protobuf_config/src/vm_runner.rs
+++ b/core/lib/protobuf_config/src/vm_runner.rs
@@ -1,6 +1,6 @@
 use anyhow::Context;
 use zksync_basic_types::L1BatchNumber;
-use zksync_config::configs::{self};
+use zksync_config::configs;
 use zksync_protobuf::{required, ProtoRepr};
 
 use crate::proto::vm_runner as proto;
diff --git a/core/lib/types/src/lib.rs b/core/lib/types/src/lib.rs
index 105d43aa6c6c..a55f6b5753db 100644
--- a/core/lib/types/src/lib.rs
+++ b/core/lib/types/src/lib.rs
@@ -16,7 +16,7 @@ pub use protocol_upgrade::{ProtocolUpgrade, ProtocolVersion};
 use serde::{Deserialize, Serialize};
 pub use storage::*;
 pub use tx::Execute;
-pub use zksync_basic_types::{protocol_version::ProtocolVersionId, vm_version::VmVersion, *};
+pub use zksync_basic_types::{protocol_version::ProtocolVersionId, vm, *};
 pub use zksync_crypto_primitives::*;
 use zksync_utils::{
     address_to_u256, bytecode::hash_bytecode, h256_to_u256, u256_to_account_address,
diff --git a/core/lib/zksync_core_leftovers/src/lib.rs b/core/lib/zksync_core_leftovers/src/lib.rs
index b79b86d718d0..9022ecd2f597 100644
--- a/core/lib/zksync_core_leftovers/src/lib.rs
+++ b/core/lib/zksync_core_leftovers/src/lib.rs
@@ -62,6 +62,8 @@ pub enum Component {
     BaseTokenRatioPersister,
     /// VM runner-based component that saves VM execution data for basic witness generation.
     VmRunnerBwip,
+    /// VM runner-based component that allows to test experimental VM features. Doesn't save any data to Postgres.
+    VmPlayground,
 }
 
 #[derive(Debug)]
@@ -106,6 +108,7 @@ impl FromStr for Components {
                 Ok(Components(vec![Component::BaseTokenRatioPersister]))
             }
             "vm_runner_bwip" => Ok(Components(vec![Component::VmRunnerBwip])),
+            "vm_playground" => Ok(Components(vec![Component::VmPlayground])),
             other => Err(format!("{} is not a valid component name", other)),
         }
     }
diff --git a/core/lib/zksync_core_leftovers/src/temp_config_store/mod.rs b/core/lib/zksync_core_leftovers/src/temp_config_store/mod.rs
index 1ad688ed14cb..75aad29903b5 100644
--- a/core/lib/zksync_core_leftovers/src/temp_config_store/mod.rs
+++ b/core/lib/zksync_core_leftovers/src/temp_config_store/mod.rs
@@ -12,11 +12,11 @@ use zksync_config::{
         house_keeper::HouseKeeperConfig,
         vm_runner::BasicWitnessInputProducerConfig,
         wallets::{AddressWallet, EthSender, StateKeeper, Wallet, Wallets},
-        CommitmentGeneratorConfig, DatabaseSecrets, ExternalPriceApiClientConfig,
-        FriProofCompressorConfig, FriProverConfig, FriProverGatewayConfig,
-        FriWitnessGeneratorConfig, FriWitnessVectorGeneratorConfig, GeneralConfig,
-        ObservabilityConfig, PrometheusConfig, ProofDataHandlerConfig, ProtectiveReadsWriterConfig,
-        PruningConfig, SnapshotRecoveryConfig,
+        CommitmentGeneratorConfig, DatabaseSecrets, ExperimentalVmPlaygroundConfig,
+        ExternalPriceApiClientConfig, FriProofCompressorConfig, FriProverConfig,
+        FriProverGatewayConfig, FriWitnessGeneratorConfig, FriWitnessVectorGeneratorConfig,
+        GeneralConfig, ObservabilityConfig, PrometheusConfig, ProofDataHandlerConfig,
+        ProtectiveReadsWriterConfig, PruningConfig, SnapshotRecoveryConfig,
     },
     ApiConfig, BaseTokenAdjusterConfig, ContractVerifierConfig, DADispatcherConfig, DBConfig,
     EthConfig, EthWatchConfig, GasAdjusterConfig, ObjectStoreConfig, PostgresConfig,
@@ -71,6 +71,7 @@ pub struct TempConfigStore {
     pub da_dispatcher_config: Option<DADispatcherConfig>,
     pub protective_reads_writer_config: Option<ProtectiveReadsWriterConfig>,
     pub basic_witness_input_producer_config: Option<BasicWitnessInputProducerConfig>,
+    pub vm_playground_config: Option<ExperimentalVmPlaygroundConfig>,
     pub core_object_store: Option<ObjectStoreConfig>,
     pub base_token_adjuster_config: Option<BaseTokenAdjusterConfig>,
     pub commitment_generator: Option<CommitmentGeneratorConfig>,
@@ -105,6 +106,7 @@ impl TempConfigStore {
             da_dispatcher_config: self.da_dispatcher_config.clone(),
             protective_reads_writer_config: self.protective_reads_writer_config.clone(),
             basic_witness_input_producer_config: self.basic_witness_input_producer_config.clone(),
+            vm_playground_config: self.vm_playground_config.clone(),
             core_object_store: self.core_object_store.clone(),
             base_token_adjuster: self.base_token_adjuster_config.clone(),
             commitment_generator: self.commitment_generator.clone(),
@@ -177,6 +179,7 @@ fn load_env_config() -> anyhow::Result<TempConfigStore> {
         da_dispatcher_config: DADispatcherConfig::from_env().ok(),
         protective_reads_writer_config: ProtectiveReadsWriterConfig::from_env().ok(),
         basic_witness_input_producer_config: BasicWitnessInputProducerConfig::from_env().ok(),
+        vm_playground_config: ExperimentalVmPlaygroundConfig::from_env().ok(),
         core_object_store: ObjectStoreConfig::from_env().ok(),
         base_token_adjuster_config: BaseTokenAdjusterConfig::from_env().ok(),
         commitment_generator: None,
@@ -186,7 +189,7 @@ fn load_env_config() -> anyhow::Result<TempConfigStore> {
     })
 }
 
-pub fn load_general_config(path: Option<std::path::PathBuf>) -> anyhow::Result<GeneralConfig> {
+pub fn load_general_config(path: Option<PathBuf>) -> anyhow::Result<GeneralConfig> {
     match path {
         Some(path) => {
             let yaml = std::fs::read_to_string(path).context("Failed to read general config")?;
@@ -198,7 +201,7 @@ pub fn load_general_config(path: Option<std::path::PathBuf>) -> anyhow::Result<G
     }
 }
 
-pub fn load_database_secrets(path: Option<std::path::PathBuf>) -> anyhow::Result<DatabaseSecrets> {
+pub fn load_database_secrets(path: Option<PathBuf>) -> anyhow::Result<DatabaseSecrets> {
     match path {
         Some(path) => {
             let yaml = std::fs::read_to_string(path).context("Failed to read secrets")?;
diff --git a/core/node/api_server/src/tx_sender/mod.rs b/core/node/api_server/src/tx_sender/mod.rs
index 38939937fcda..826200b5537c 100644
--- a/core/node/api_server/src/tx_sender/mod.rs
+++ b/core/node/api_server/src/tx_sender/mod.rs
@@ -31,9 +31,9 @@ use zksync_types::{
     l2::{error::TxCheckError::TxDuplication, L2Tx},
     transaction_request::CallOverrides,
     utils::storage_key_for_eth_balance,
+    vm::VmVersion,
     AccountTreeId, Address, ExecuteTransactionCommon, L2ChainId, Nonce, PackedEthSignature,
-    ProtocolVersionId, Transaction, VmVersion, H160, H256, MAX_L2_TX_GAS_LIMIT,
-    MAX_NEW_FACTORY_DEPS, U256,
+    ProtocolVersionId, Transaction, H160, H256, MAX_L2_TX_GAS_LIMIT, MAX_NEW_FACTORY_DEPS, U256,
 };
 use zksync_utils::h256_to_u256;
 
diff --git a/core/node/commitment_generator/src/utils.rs b/core/node/commitment_generator/src/utils.rs
index b4e6bc542e97..59f8753859a4 100644
--- a/core/node/commitment_generator/src/utils.rs
+++ b/core/node/commitment_generator/src/utils.rs
@@ -15,7 +15,7 @@ use zk_evm_1_5_0::{
     zk_evm_abstractions::queries::LogQuery as LogQuery_1_5_0,
 };
 use zksync_multivm::utils::get_used_bootloader_memory_bytes;
-use zksync_types::{zk_evm_types::LogQuery, ProtocolVersionId, VmVersion, H256, U256};
+use zksync_types::{vm::VmVersion, zk_evm_types::LogQuery, ProtocolVersionId, H256, U256};
 use zksync_utils::expand_memory_contents;
 
 /// Encapsulates computations of commitment components.
diff --git a/core/node/node_framework/src/implementations/layers/state_keeper/main_batch_executor.rs b/core/node/node_framework/src/implementations/layers/state_keeper/main_batch_executor.rs
index 33d3b5676aac..91b89adc20af 100644
--- a/core/node/node_framework/src/implementations/layers/state_keeper/main_batch_executor.rs
+++ b/core/node/node_framework/src/implementations/layers/state_keeper/main_batch_executor.rs
@@ -3,7 +3,6 @@ use zksync_state_keeper::MainBatchExecutor;
 use crate::{
     implementations::resources::state_keeper::BatchExecutorResource,
     wiring_layer::{WiringError, WiringLayer},
-    IntoContext,
 };
 
 /// Wiring layer for `MainBatchExecutor`, part of the state keeper responsible for running the VM.
@@ -13,12 +12,6 @@ pub struct MainBatchExecutorLayer {
     optional_bytecode_compression: bool,
 }
 
-#[derive(Debug, IntoContext)]
-#[context(crate = crate)]
-pub struct Output {
-    pub batch_executor: BatchExecutorResource,
-}
-
 impl MainBatchExecutorLayer {
     pub fn new(save_call_traces: bool, optional_bytecode_compression: bool) -> Self {
         Self {
@@ -31,18 +24,15 @@ impl MainBatchExecutorLayer {
 #[async_trait::async_trait]
 impl WiringLayer for MainBatchExecutorLayer {
     type Input = ();
-    type Output = Output;
+    type Output = BatchExecutorResource;
 
     fn layer_name(&self) -> &'static str {
         "main_batch_executor_layer"
     }
 
-    async fn wire(self, _input: Self::Input) -> Result<Self::Output, WiringError> {
-        let builder =
+    async fn wire(self, (): Self::Input) -> Result<Self::Output, WiringError> {
+        let executor =
             MainBatchExecutor::new(self.save_call_traces, self.optional_bytecode_compression);
-
-        Ok(Output {
-            batch_executor: builder.into(),
-        })
+        Ok(executor.into())
     }
 }
diff --git a/core/node/node_framework/src/implementations/layers/vm_runner/bwip.rs b/core/node/node_framework/src/implementations/layers/vm_runner/bwip.rs
index 74b4b5e32072..ee2fb84416e1 100644
--- a/core/node/node_framework/src/implementations/layers/vm_runner/bwip.rs
+++ b/core/node/node_framework/src/implementations/layers/vm_runner/bwip.rs
@@ -1,8 +1,9 @@
 use zksync_config::configs::vm_runner::BasicWitnessInputProducerConfig;
+use zksync_state_keeper::MainBatchExecutor;
 use zksync_types::L2ChainId;
 use zksync_vm_runner::{
-    BasicWitnessInputProducer, BasicWitnessInputProducerIo, ConcurrentOutputHandlerFactoryTask,
-    StorageSyncTask,
+    impls::{BasicWitnessInputProducer, BasicWitnessInputProducerIo},
+    ConcurrentOutputHandlerFactoryTask, StorageSyncTask,
 };
 
 use crate::{
@@ -18,17 +19,14 @@ use crate::{
 
 #[derive(Debug)]
 pub struct BasicWitnessInputProducerLayer {
-    basic_witness_input_producer_config: BasicWitnessInputProducerConfig,
+    config: BasicWitnessInputProducerConfig,
     zksync_network_id: L2ChainId,
 }
 
 impl BasicWitnessInputProducerLayer {
-    pub fn new(
-        basic_witness_input_producer_config: BasicWitnessInputProducerConfig,
-        zksync_network_id: L2ChainId,
-    ) -> Self {
+    pub fn new(config: BasicWitnessInputProducerConfig, zksync_network_id: L2ChainId) -> Self {
         Self {
-            basic_witness_input_producer_config,
+            config,
             zksync_network_id,
         }
     }
@@ -68,25 +66,26 @@ impl WiringLayer for BasicWitnessInputProducerLayer {
             object_store,
         } = input;
 
+        // - 1 connection for `StorageSyncTask` which can hold a long-term connection in case it needs to
+        //   catch up cache.
+        // - 1 connection for `ConcurrentOutputHandlerFactoryTask` / `VmRunner` as they need occasional access
+        //   to DB for querying last processed batch and last ready to be loaded batch.
+        // - `window_size` connections for `BasicWitnessInputProducer`
+        //   as there can be multiple output handlers holding multi-second connections to process
+        //   BWIP data.
+        let connection_pool = master_pool.get_custom(self.config.window_size + 2).await?;
+
+        // We don't get the executor from the context because it would contain state keeper-specific settings.
+        let batch_executor = Box::new(MainBatchExecutor::new(false, false));
+
         let (basic_witness_input_producer, tasks) = BasicWitnessInputProducer::new(
-            // One for `StorageSyncTask` which can hold a long-term connection in case it needs to
-            // catch up cache.
-            //
-            // One for `ConcurrentOutputHandlerFactoryTask`/`VmRunner` as they need occasional access
-            // to DB for querying last processed batch and last ready to be loaded batch.
-            //
-            // `window_size` connections for `BasicWitnessInputProducer`
-            // as there can be multiple output handlers holding multi-second connections to process
-            // BWIP data.
-            master_pool
-                .get_custom(self.basic_witness_input_producer_config.window_size + 2)
-                .await?,
+            connection_pool,
             object_store.0,
-            self.basic_witness_input_producer_config.db_path,
+            batch_executor,
+            self.config.db_path,
             self.zksync_network_id,
-            self.basic_witness_input_producer_config
-                .first_processed_batch,
-            self.basic_witness_input_producer_config.window_size,
+            self.config.first_processed_batch,
+            self.config.window_size,
         )
         .await?;
 
diff --git a/core/node/node_framework/src/implementations/layers/vm_runner/mod.rs b/core/node/node_framework/src/implementations/layers/vm_runner/mod.rs
index 91e92ffcd1ba..85b7028bc799 100644
--- a/core/node/node_framework/src/implementations/layers/vm_runner/mod.rs
+++ b/core/node/node_framework/src/implementations/layers/vm_runner/mod.rs
@@ -6,6 +6,7 @@ use crate::{
 };
 
 pub mod bwip;
+pub mod playground;
 pub mod protective_reads;
 
 #[async_trait::async_trait]
diff --git a/core/node/node_framework/src/implementations/layers/vm_runner/playground.rs b/core/node/node_framework/src/implementations/layers/vm_runner/playground.rs
new file mode 100644
index 000000000000..549a3ed13e49
--- /dev/null
+++ b/core/node/node_framework/src/implementations/layers/vm_runner/playground.rs
@@ -0,0 +1,108 @@
+use async_trait::async_trait;
+use zksync_config::configs::ExperimentalVmPlaygroundConfig;
+use zksync_node_framework_derive::{FromContext, IntoContext};
+use zksync_state_keeper::MainBatchExecutor;
+use zksync_types::L2ChainId;
+use zksync_vm_runner::{
+    impls::{VmPlayground, VmPlaygroundIo, VmPlaygroundLoaderTask},
+    ConcurrentOutputHandlerFactoryTask,
+};
+
+use crate::{
+    implementations::resources::pools::{MasterPool, PoolResource},
+    StopReceiver, Task, TaskId, WiringError, WiringLayer,
+};
+
+#[derive(Debug)]
+pub struct VmPlaygroundLayer {
+    config: ExperimentalVmPlaygroundConfig,
+    zksync_network_id: L2ChainId,
+}
+
+impl VmPlaygroundLayer {
+    pub fn new(config: ExperimentalVmPlaygroundConfig, zksync_network_id: L2ChainId) -> Self {
+        Self {
+            config,
+            zksync_network_id,
+        }
+    }
+}
+
+#[derive(Debug, FromContext)]
+#[context(crate = crate)]
+pub struct Input {
+    pub master_pool: PoolResource<MasterPool>,
+}
+
+#[derive(Debug, IntoContext)]
+#[context(crate = crate)]
+pub struct Output {
+    #[context(task)]
+    pub output_handler_factory_task: ConcurrentOutputHandlerFactoryTask<VmPlaygroundIo>,
+    #[context(task)]
+    pub loader_task: VmPlaygroundLoaderTask,
+    #[context(task)]
+    pub playground: VmPlayground,
+}
+
+#[async_trait]
+impl WiringLayer for VmPlaygroundLayer {
+    type Input = Input;
+    type Output = Output;
+
+    fn layer_name(&self) -> &'static str {
+        "vm_runner_playground"
+    }
+
+    async fn wire(self, input: Self::Input) -> Result<Self::Output, WiringError> {
+        let Input { master_pool } = input;
+
+        // - 1 connection for `StorageSyncTask` which can hold a long-term connection in case it needs to
+        //   catch up cache.
+        // - 1 connection for `ConcurrentOutputHandlerFactoryTask` / `VmRunner` as they need occasional access
+        //   to DB for querying last processed batch and last ready to be loaded batch.
+        // - 1 connection for the only running VM instance.
+        let connection_pool = master_pool.get_custom(3).await?;
+
+        let mut batch_executor = Box::new(MainBatchExecutor::new(false, false));
+        batch_executor.set_fast_vm_mode(self.config.fast_vm_mode);
+
+        let (playground, tasks) = VmPlayground::new(
+            connection_pool,
+            batch_executor,
+            self.config.db_path,
+            self.zksync_network_id,
+            self.config.first_processed_batch,
+            self.config.reset,
+        )
+        .await?;
+
+        Ok(Output {
+            output_handler_factory_task: tasks.output_handler_factory_task,
+            loader_task: tasks.loader_task,
+            playground,
+        })
+    }
+}
+
+#[async_trait]
+impl Task for VmPlaygroundLoaderTask {
+    fn id(&self) -> TaskId {
+        "vm_runner/playground/storage_sync".into()
+    }
+
+    async fn run(self: Box<Self>, stop_receiver: StopReceiver) -> anyhow::Result<()> {
+        (*self).run(stop_receiver.0).await
+    }
+}
+
+#[async_trait]
+impl Task for VmPlayground {
+    fn id(&self) -> TaskId {
+        "vm_runner/playground".into()
+    }
+
+    async fn run(self: Box<Self>, stop_receiver: StopReceiver) -> anyhow::Result<()> {
+        (*self).run(&stop_receiver.0).await
+    }
+}
diff --git a/core/node/node_framework/src/implementations/layers/vm_runner/protective_reads.rs b/core/node/node_framework/src/implementations/layers/vm_runner/protective_reads.rs
index 3b07d0cea139..a0b0d18a4d93 100644
--- a/core/node/node_framework/src/implementations/layers/vm_runner/protective_reads.rs
+++ b/core/node/node_framework/src/implementations/layers/vm_runner/protective_reads.rs
@@ -2,7 +2,8 @@ use zksync_config::configs::vm_runner::ProtectiveReadsWriterConfig;
 use zksync_node_framework_derive::FromContext;
 use zksync_types::L2ChainId;
 use zksync_vm_runner::{
-    ConcurrentOutputHandlerFactoryTask, ProtectiveReadsIo, ProtectiveReadsWriter, StorageSyncTask,
+    impls::{ProtectiveReadsIo, ProtectiveReadsWriter},
+    ConcurrentOutputHandlerFactoryTask, StorageSyncTask,
 };
 
 use crate::{
diff --git a/core/node/state_keeper/src/batch_executor/main_executor.rs b/core/node/state_keeper/src/batch_executor/main_executor.rs
index e96cf9f1f7cb..7838ef9c4b59 100644
--- a/core/node/state_keeper/src/batch_executor/main_executor.rs
+++ b/core/node/state_keeper/src/batch_executor/main_executor.rs
@@ -10,7 +10,7 @@ use tokio::{
 use zksync_multivm::{
     interface::{
         ExecutionResult, FinishedL1Batch, Halt, L1BatchEnv, L2BlockEnv, SystemEnv,
-        VmExecutionResultAndLogs, VmFactory, VmInterface, VmInterfaceHistoryEnabled,
+        VmExecutionResultAndLogs, VmInterface, VmInterfaceHistoryEnabled,
     },
     tracers::CallTracer,
     vm_latest::HistoryEnabled,
@@ -18,7 +18,7 @@ use zksync_multivm::{
 };
 use zksync_shared_metrics::{InteractionType, TxStage, APP_METRICS};
 use zksync_state::{ReadStorage, ReadStorageFactory, StorageView};
-use zksync_types::{vm_trace::Call, Transaction};
+use zksync_types::{vm::FastVmMode, vm_trace::Call, Transaction};
 use zksync_utils::bytecode::CompressedBytecodeInfo;
 
 use super::{BatchExecutor, BatchExecutorHandle, Command, TxExecutionResult};
@@ -39,6 +39,7 @@ pub struct MainBatchExecutor {
     /// that in cases where the node is expected to process any transactions processed by the sequencer
     /// regardless of its configuration, this flag should be set to `true`.
     optional_bytecode_compression: bool,
+    fast_vm_mode: FastVmMode,
 }
 
 impl MainBatchExecutor {
@@ -46,8 +47,18 @@ impl MainBatchExecutor {
         Self {
             save_call_traces,
             optional_bytecode_compression,
+            fast_vm_mode: FastVmMode::Old,
         }
     }
+
+    pub fn set_fast_vm_mode(&mut self, fast_vm_mode: FastVmMode) {
+        if !matches!(fast_vm_mode, FastVmMode::Old) {
+            tracing::warn!(
+                "Running new VM with mode {fast_vm_mode:?}; this can lead to incorrect node behavior"
+            );
+        }
+        self.fast_vm_mode = fast_vm_mode;
+    }
 }
 
 #[async_trait]
@@ -65,6 +76,7 @@ impl BatchExecutor for MainBatchExecutor {
         let executor = CommandReceiver {
             save_call_traces: self.save_call_traces,
             optional_bytecode_compression: self.optional_bytecode_compression,
+            fast_vm_mode: self.fast_vm_mode,
             commands: commands_receiver,
         };
 
@@ -96,6 +108,7 @@ impl BatchExecutor for MainBatchExecutor {
 struct CommandReceiver {
     save_call_traces: bool,
     optional_bytecode_compression: bool,
+    fast_vm_mode: FastVmMode,
     commands: mpsc::Receiver<Command>,
 }
 
@@ -109,8 +122,12 @@ impl CommandReceiver {
         tracing::info!("Starting executing L1 batch #{}", &l1_batch_params.number);
 
         let storage_view = StorageView::new(secondary_storage).to_rc_ptr();
-
-        let mut vm = VmInstance::new(l1_batch_params, system_env, storage_view.clone());
+        let mut vm = VmInstance::maybe_fast(
+            l1_batch_params,
+            system_env,
+            storage_view.clone(),
+            self.fast_vm_mode,
+        );
 
         while let Some(cmd) = self.commands.blocking_recv() {
             match cmd {
diff --git a/core/node/state_keeper/src/batch_executor/tests/mod.rs b/core/node/state_keeper/src/batch_executor/tests/mod.rs
index 4b36965895fd..ab9115991deb 100644
--- a/core/node/state_keeper/src/batch_executor/tests/mod.rs
+++ b/core/node/state_keeper/src/batch_executor/tests/mod.rs
@@ -2,7 +2,9 @@ use assert_matches::assert_matches;
 use test_casing::{test_casing, Product};
 use zksync_dal::{ConnectionPool, Core};
 use zksync_test_account::Account;
-use zksync_types::{get_nonce_key, utils::storage_key_for_eth_balance, PriorityOpId};
+use zksync_types::{
+    get_nonce_key, utils::storage_key_for_eth_balance, vm::FastVmMode, PriorityOpId,
+};
 
 use self::tester::{AccountLoadNextExecutable, StorageSnapshot, TestConfig, Tester};
 use super::TxExecutionResult;
@@ -41,13 +43,15 @@ impl StorageType {
     const ALL: [Self; 3] = [Self::AsyncRocksdbCache, Self::Rocksdb, Self::Postgres];
 }
 
+const FAST_VM_MODES: [FastVmMode; 3] = [FastVmMode::Old, FastVmMode::New, FastVmMode::Shadow];
+
 /// Checks that we can successfully execute a single L2 tx in batch executor on all storage types.
-#[test_casing(3, StorageType::ALL)]
+#[test_casing(9, Product((StorageType::ALL, FAST_VM_MODES)))]
 #[tokio::test]
-async fn execute_l2_tx(storage_type: StorageType) {
+async fn execute_l2_tx(storage_type: StorageType, vm_mode: FastVmMode) {
     let connection_pool = ConnectionPool::<Core>::constrained_test_pool(1).await;
     let mut alice = Account::random();
-    let mut tester = Tester::new(connection_pool);
+    let mut tester = Tester::new(connection_pool, vm_mode);
     tester.genesis().await;
     tester.fund(&[alice.address()]).await;
     let mut executor = tester.create_batch_executor(storage_type).await;
@@ -82,14 +86,9 @@ impl SnapshotRecoveryMutation {
     }
 }
 
-const EXECUTE_L2_TX_AFTER_SNAPSHOT_RECOVERY_CASES: Product<(
-    [Option<SnapshotRecoveryMutation>; 3],
-    [StorageType; 3],
-)> = Product((SnapshotRecoveryMutation::ALL, StorageType::ALL));
-
 /// Tests that we can continue executing account transactions after emulating snapshot recovery.
 /// Test cases with a set `mutation` ensure that the VM executor correctly detects missing data (e.g., dropped account nonce).
-#[test_casing(9, EXECUTE_L2_TX_AFTER_SNAPSHOT_RECOVERY_CASES)]
+#[test_casing(9, Product((SnapshotRecoveryMutation::ALL, StorageType::ALL)))]
 #[tokio::test]
 async fn execute_l2_tx_after_snapshot_recovery(
     mutation: Option<SnapshotRecoveryMutation>,
@@ -106,7 +105,7 @@ async fn execute_l2_tx_after_snapshot_recovery(
     }
     let snapshot = storage_snapshot.recover(&connection_pool).await;
 
-    let mut tester = Tester::new(connection_pool);
+    let mut tester = Tester::new(connection_pool, FastVmMode::Old);
     let mut executor = tester
         .recover_batch_executor_custom(&storage_type, &snapshot)
         .await;
@@ -120,12 +119,13 @@ async fn execute_l2_tx_after_snapshot_recovery(
 }
 
 /// Checks that we can successfully execute a single L1 tx in batch executor.
+#[test_casing(3, FAST_VM_MODES)]
 #[tokio::test]
-async fn execute_l1_tx() {
+async fn execute_l1_tx(vm_mode: FastVmMode) {
     let connection_pool = ConnectionPool::<Core>::constrained_test_pool(1).await;
     let mut alice = Account::random();
 
-    let mut tester = Tester::new(connection_pool);
+    let mut tester = Tester::new(connection_pool, vm_mode);
 
     tester.genesis().await;
     tester.fund(&[alice.address()]).await;
@@ -142,12 +142,13 @@ async fn execute_l1_tx() {
 }
 
 /// Checks that we can successfully execute a single L2 tx and a single L1 tx in batch executor.
+#[test_casing(3, FAST_VM_MODES)]
 #[tokio::test]
-async fn execute_l2_and_l1_txs() {
+async fn execute_l2_and_l1_txs(vm_mode: FastVmMode) {
     let connection_pool = ConnectionPool::<Core>::constrained_test_pool(1).await;
     let mut alice = Account::random();
 
-    let mut tester = Tester::new(connection_pool);
+    let mut tester = Tester::new(connection_pool, vm_mode);
     tester.genesis().await;
     tester.fund(&[alice.address()]).await;
     let mut executor = tester
@@ -167,12 +168,13 @@ async fn execute_l2_and_l1_txs() {
 }
 
 /// Checks that we can successfully rollback the transaction and execute it once again.
+#[test_casing(3, FAST_VM_MODES)]
 #[tokio::test]
-async fn rollback() {
+async fn rollback(vm_mode: FastVmMode) {
     let connection_pool = ConnectionPool::<Core>::constrained_test_pool(1).await;
     let mut alice = Account::random();
 
-    let mut tester = Tester::new(connection_pool);
+    let mut tester = Tester::new(connection_pool, vm_mode);
 
     tester.genesis().await;
     tester.fund(&[alice.address()]).await;
@@ -213,12 +215,13 @@ async fn rollback() {
 }
 
 /// Checks that incorrect transactions are marked as rejected.
+#[test_casing(3, FAST_VM_MODES)]
 #[tokio::test]
-async fn reject_tx() {
+async fn reject_tx(vm_mode: FastVmMode) {
     let connection_pool = ConnectionPool::<Core>::constrained_test_pool(1).await;
     let mut alice = Account::random();
 
-    let mut tester = Tester::new(connection_pool);
+    let mut tester = Tester::new(connection_pool, vm_mode);
 
     tester.genesis().await;
     let mut executor = tester
@@ -232,12 +235,13 @@ async fn reject_tx() {
 
 /// Checks that tx with too big gas limit is correctly processed.
 /// When processed in the bootloader, no more than 80M gas can be used within the execution context.
+#[test_casing(3, FAST_VM_MODES)]
 #[tokio::test]
-async fn too_big_gas_limit() {
+async fn too_big_gas_limit(vm_mode: FastVmMode) {
     let connection_pool = ConnectionPool::<Core>::constrained_test_pool(1).await;
     let mut alice = Account::random();
 
-    let mut tester = Tester::new(connection_pool);
+    let mut tester = Tester::new(connection_pool, vm_mode);
     tester.genesis().await;
     tester.fund(&[alice.address()]).await;
     let mut executor = tester
@@ -252,12 +256,13 @@ async fn too_big_gas_limit() {
 }
 
 /// Checks that we can't execute the same transaction twice.
+#[test_casing(3, FAST_VM_MODES)]
 #[tokio::test]
-async fn tx_cant_be_reexecuted() {
+async fn tx_cant_be_reexecuted(vm_mode: FastVmMode) {
     let connection_pool = ConnectionPool::<Core>::constrained_test_pool(1).await;
     let mut alice = Account::random();
 
-    let mut tester = Tester::new(connection_pool);
+    let mut tester = Tester::new(connection_pool, vm_mode);
     tester.genesis().await;
     tester.fund(&[alice.address()]).await;
     let mut executor = tester
@@ -274,12 +279,13 @@ async fn tx_cant_be_reexecuted() {
 }
 
 /// Checks that we can deploy and call the loadnext contract.
+#[test_casing(3, FAST_VM_MODES)]
 #[tokio::test]
-async fn deploy_and_call_loadtest() {
+async fn deploy_and_call_loadtest(vm_mode: FastVmMode) {
     let connection_pool = ConnectionPool::<Core>::constrained_test_pool(1).await;
     let mut alice = Account::random();
 
-    let mut tester = Tester::new(connection_pool);
+    let mut tester = Tester::new(connection_pool, vm_mode);
     tester.genesis().await;
     tester.fund(&[alice.address()]).await;
     let mut executor = tester
@@ -304,12 +310,13 @@ async fn deploy_and_call_loadtest() {
 }
 
 /// Checks that a tx that is reverted by the VM still can be included into a batch.
+#[test_casing(3, FAST_VM_MODES)]
 #[tokio::test]
-async fn execute_reverted_tx() {
+async fn execute_reverted_tx(vm_mode: FastVmMode) {
     let connection_pool = ConnectionPool::<Core>::constrained_test_pool(1).await;
     let mut alice = Account::random();
 
-    let mut tester = Tester::new(connection_pool);
+    let mut tester = Tester::new(connection_pool, vm_mode);
 
     tester.genesis().await;
     tester.fund(&[alice.address()]).await;
@@ -334,13 +341,14 @@ async fn execute_reverted_tx() {
 
 /// Runs the batch executor through a semi-realistic basic scenario:
 /// a batch with different operations, both successful and not.
+#[test_casing(3, FAST_VM_MODES)]
 #[tokio::test]
-async fn execute_realistic_scenario() {
+async fn execute_realistic_scenario(vm_mode: FastVmMode) {
     let connection_pool = ConnectionPool::<Core>::constrained_test_pool(1).await;
     let mut alice = Account::random();
     let mut bob = Account::random();
 
-    let mut tester = Tester::new(connection_pool);
+    let mut tester = Tester::new(connection_pool, vm_mode);
 
     tester.genesis().await;
     tester.fund(&[alice.address()]).await;
@@ -395,8 +403,9 @@ async fn execute_realistic_scenario() {
 }
 
 /// Checks that we handle the bootloader out of gas error on execution phase.
+#[test_casing(3, FAST_VM_MODES)]
 #[tokio::test]
-async fn bootloader_out_of_gas_for_any_tx() {
+async fn bootloader_out_of_gas_for_any_tx(vm_mode: FastVmMode) {
     let connection_pool = ConnectionPool::<Core>::constrained_test_pool(1).await;
     let mut alice = Account::random();
 
@@ -406,6 +415,7 @@ async fn bootloader_out_of_gas_for_any_tx() {
             save_call_traces: false,
             vm_gas_limit: Some(10),
             validation_computational_gas_limit: u32::MAX,
+            fast_vm_mode: vm_mode,
         },
     );
 
@@ -426,7 +436,7 @@ async fn bootloader_tip_out_of_gas() {
     let connection_pool = ConnectionPool::<Core>::constrained_test_pool(1).await;
     let mut alice = Account::random();
 
-    let mut tester = Tester::new(connection_pool);
+    let mut tester = Tester::new(connection_pool, FastVmMode::Old);
 
     tester.genesis().await;
     tester.fund(&[alice.address()]).await;
@@ -451,6 +461,7 @@ async fn bootloader_tip_out_of_gas() {
                 - 10,
         ),
         validation_computational_gas_limit: u32::MAX,
+        fast_vm_mode: FastVmMode::Old,
     });
 
     let mut second_executor = tester
@@ -467,7 +478,7 @@ async fn catchup_rocksdb_cache() {
     let mut alice = Account::random();
     let mut bob = Account::random();
 
-    let mut tester = Tester::new(connection_pool);
+    let mut tester = Tester::new(connection_pool, FastVmMode::Old);
 
     tester.genesis().await;
     tester.fund(&[alice.address(), bob.address()]).await;
diff --git a/core/node/state_keeper/src/batch_executor/tests/tester.rs b/core/node/state_keeper/src/batch_executor/tests/tester.rs
index 579f3bee4819..f4e7d30d0714 100644
--- a/core/node/state_keeper/src/batch_executor/tests/tester.rs
+++ b/core/node/state_keeper/src/batch_executor/tests/tester.rs
@@ -24,6 +24,7 @@ use zksync_types::{
     storage_writes_deduplicator::StorageWritesDeduplicator,
     system_contracts::get_system_smart_contracts,
     utils::storage_key_for_standard_token_balance,
+    vm::FastVmMode,
     AccountTreeId, Address, Execute, L1BatchNumber, L2BlockNumber, PriorityOpId, ProtocolVersionId,
     StorageLog, Transaction, H256, L2_BASE_TOKEN_ADDRESS, U256,
 };
@@ -48,16 +49,18 @@ pub(super) struct TestConfig {
     pub(super) save_call_traces: bool,
     pub(super) vm_gas_limit: Option<u32>,
     pub(super) validation_computational_gas_limit: u32,
+    pub(super) fast_vm_mode: FastVmMode,
 }
 
 impl TestConfig {
-    pub(super) fn new() -> Self {
+    pub(super) fn new(fast_vm_mode: FastVmMode) -> Self {
         let config = StateKeeperConfig::for_tests();
 
         Self {
             vm_gas_limit: None,
             save_call_traces: false,
             validation_computational_gas_limit: config.validation_computational_gas_limit,
+            fast_vm_mode,
         }
     }
 }
@@ -74,8 +77,8 @@ pub(super) struct Tester {
 }
 
 impl Tester {
-    pub(super) fn new(pool: ConnectionPool<Core>) -> Self {
-        Self::with_config(pool, TestConfig::new())
+    pub(super) fn new(pool: ConnectionPool<Core>, fast_vm_mode: FastVmMode) -> Self {
+        Self::with_config(pool, TestConfig::new(fast_vm_mode))
     }
 
     pub(super) fn with_config(pool: ConnectionPool<Core>, config: TestConfig) -> Self {
@@ -148,6 +151,8 @@ impl Tester {
         system_env: SystemEnv,
     ) -> BatchExecutorHandle {
         let mut batch_executor = MainBatchExecutor::new(self.config.save_call_traces, false);
+        batch_executor.set_fast_vm_mode(self.config.fast_vm_mode);
+
         let (_stop_sender, stop_receiver) = watch::channel(false);
         batch_executor
             .init_batch(storage_factory, l1_batch_env, system_env, &stop_receiver)
@@ -448,7 +453,7 @@ impl StorageSnapshot {
         alice: &mut Account,
         transaction_count: u32,
     ) -> Self {
-        let mut tester = Tester::new(connection_pool.clone());
+        let mut tester = Tester::new(connection_pool.clone(), FastVmMode::Old);
         tester.genesis().await;
         tester.fund(&[alice.address()]).await;
 
diff --git a/core/node/state_keeper/src/mempool_actor.rs b/core/node/state_keeper/src/mempool_actor.rs
index d79d9ebb34a8..5003d75b6694 100644
--- a/core/node/state_keeper/src/mempool_actor.rs
+++ b/core/node/state_keeper/src/mempool_actor.rs
@@ -11,7 +11,7 @@ use zksync_multivm::utils::derive_base_fee_and_gas_per_pubdata;
 use zksync_node_fee_model::BatchFeeModelInputProvider;
 #[cfg(test)]
 use zksync_types::H256;
-use zksync_types::{get_nonce_key, Address, Nonce, Transaction, VmVersion};
+use zksync_types::{get_nonce_key, vm::VmVersion, Address, Nonce, Transaction};
 
 use super::{metrics::KEEPER_METRICS, types::MempoolGuard};
 
diff --git a/core/node/vm_runner/Cargo.toml b/core/node/vm_runner/Cargo.toml
index 3af52ed4688e..52a8e4676437 100644
--- a/core/node/vm_runner/Cargo.toml
+++ b/core/node/vm_runner/Cargo.toml
@@ -39,3 +39,4 @@ backon.workspace = true
 futures = { workspace = true, features = ["compat"] }
 rand.workspace = true
 tempfile.workspace = true
+test-casing.workspace = true
diff --git a/core/node/vm_runner/src/impls/bwip.rs b/core/node/vm_runner/src/impls/bwip.rs
index 7ab18397353d..8fdd85e9abe4 100644
--- a/core/node/vm_runner/src/impls/bwip.rs
+++ b/core/node/vm_runner/src/impls/bwip.rs
@@ -6,7 +6,7 @@ use tokio::sync::watch;
 use zksync_dal::{Connection, ConnectionPool, Core, CoreDal};
 use zksync_object_store::ObjectStore;
 use zksync_prover_interface::inputs::VMRunWitnessInputData;
-use zksync_state_keeper::{MainBatchExecutor, StateKeeperOutputHandler, UpdatesManager};
+use zksync_state_keeper::{BatchExecutor, StateKeeperOutputHandler, UpdatesManager};
 use zksync_types::{
     block::StorageOracleInfo, witness_block_state::WitnessStorageState, L1BatchNumber, L2ChainId,
     H256,
@@ -30,6 +30,7 @@ impl BasicWitnessInputProducer {
     pub async fn new(
         pool: ConnectionPool<Core>,
         object_store: Arc<dyn ObjectStore>,
+        batch_executor: Box<dyn BatchExecutor>,
         rocksdb_path: String,
         chain_id: L2ChainId,
         first_processed_batch: L1BatchNumber,
@@ -47,13 +48,12 @@ impl BasicWitnessInputProducer {
         };
         let (output_handler_factory, output_handler_factory_task) =
             ConcurrentOutputHandlerFactory::new(pool.clone(), io.clone(), output_handler_factory);
-        let batch_processor = MainBatchExecutor::new(false, false);
         let vm_runner = VmRunner::new(
             pool,
             Box::new(io),
             Arc::new(loader),
             Box::new(output_handler_factory),
-            Box::new(batch_processor),
+            batch_executor,
         );
         Ok((
             Self { vm_runner },
@@ -75,8 +75,7 @@ impl BasicWitnessInputProducer {
     }
 }
 
-/// A collections of tasks that need to be run in order for BWIP to work as
-/// intended.
+/// Collection of tasks that need to be run in order for BWIP to work as intended.
 #[derive(Debug)]
 pub struct BasicWitnessInputProducerTasks {
     /// Task that synchronizes storage with new available batches.
diff --git a/core/node/vm_runner/src/impls/mod.rs b/core/node/vm_runner/src/impls/mod.rs
index 2d982730498a..7f9869531c65 100644
--- a/core/node/vm_runner/src/impls/mod.rs
+++ b/core/node/vm_runner/src/impls/mod.rs
@@ -1,7 +1,13 @@
+//! Components powered by a VM runner.
+
 mod bwip;
+mod playground;
 mod protective_reads;
 
-pub use bwip::{
-    BasicWitnessInputProducer, BasicWitnessInputProducerIo, BasicWitnessInputProducerTasks,
+pub use self::{
+    bwip::{
+        BasicWitnessInputProducer, BasicWitnessInputProducerIo, BasicWitnessInputProducerTasks,
+    },
+    playground::{VmPlayground, VmPlaygroundIo, VmPlaygroundLoaderTask, VmPlaygroundTasks},
+    protective_reads::{ProtectiveReadsIo, ProtectiveReadsWriter, ProtectiveReadsWriterTasks},
 };
-pub use protective_reads::{ProtectiveReadsIo, ProtectiveReadsWriter, ProtectiveReadsWriterTasks};
diff --git a/core/node/vm_runner/src/impls/playground.rs b/core/node/vm_runner/src/impls/playground.rs
new file mode 100644
index 000000000000..11792ebdefdf
--- /dev/null
+++ b/core/node/vm_runner/src/impls/playground.rs
@@ -0,0 +1,294 @@
+use std::{
+    io,
+    path::{Path, PathBuf},
+    sync::Arc,
+};
+
+use anyhow::Context as _;
+use async_trait::async_trait;
+use tokio::{
+    fs,
+    sync::{oneshot, watch},
+};
+use zksync_dal::{Connection, ConnectionPool, Core, CoreDal};
+use zksync_state::RocksdbStorage;
+use zksync_state_keeper::{BatchExecutor, StateKeeperOutputHandler, UpdatesManager};
+use zksync_types::{L1BatchNumber, L2ChainId};
+
+use crate::{
+    ConcurrentOutputHandlerFactory, ConcurrentOutputHandlerFactoryTask, OutputHandlerFactory,
+    StorageSyncTask, VmRunner, VmRunnerIo, VmRunnerStorage,
+};
+
+/// Virtual machine playground. Does not persist anything in Postgres; instead, keeps an L1 batch cursor as a plain text file in the RocksDB directory
+/// (so that the playground doesn't repeatedly process same batches after a restart).
+#[derive(Debug)]
+pub struct VmPlayground {
+    pool: ConnectionPool<Core>,
+    batch_executor: Box<dyn BatchExecutor>,
+    rocksdb_path: String,
+    chain_id: L2ChainId,
+    io: VmPlaygroundIo,
+    loader_task_sender: oneshot::Sender<StorageSyncTask<VmPlaygroundIo>>,
+    output_handler_factory:
+        ConcurrentOutputHandlerFactory<VmPlaygroundIo, VmPlaygroundOutputHandler>,
+    reset_to_batch: Option<L1BatchNumber>,
+}
+
+impl VmPlayground {
+    /// Creates a new playground.
+    pub async fn new(
+        pool: ConnectionPool<Core>,
+        batch_executor: Box<dyn BatchExecutor>,
+        rocksdb_path: String,
+        chain_id: L2ChainId,
+        first_processed_batch: L1BatchNumber,
+        reset_state: bool,
+    ) -> anyhow::Result<(Self, VmPlaygroundTasks)> {
+        tracing::info!(
+            "Starting VM playground with executor {batch_executor:?}, first processed batch is #{first_processed_batch} \
+             (reset processing: {reset_state:?})"
+        );
+
+        let cursor_file_path = Path::new(&rocksdb_path).join("__vm_playground_cursor");
+        let latest_processed_batch = VmPlaygroundIo::read_cursor(&cursor_file_path).await?;
+        tracing::info!("Latest processed batch: {latest_processed_batch:?}");
+        let latest_processed_batch = if reset_state {
+            first_processed_batch
+        } else {
+            latest_processed_batch.unwrap_or(first_processed_batch)
+        };
+
+        let io = VmPlaygroundIo {
+            cursor_file_path,
+            latest_processed_batch: Arc::new(watch::channel(latest_processed_batch).0),
+        };
+        let (output_handler_factory, output_handler_factory_task) =
+            ConcurrentOutputHandlerFactory::new(
+                pool.clone(),
+                io.clone(),
+                VmPlaygroundOutputHandler,
+            );
+        let (loader_task_sender, loader_task_receiver) = oneshot::channel();
+
+        let this = Self {
+            pool,
+            batch_executor,
+            rocksdb_path,
+            chain_id,
+            io,
+            loader_task_sender,
+            output_handler_factory,
+            reset_to_batch: reset_state.then_some(first_processed_batch),
+        };
+        Ok((
+            this,
+            VmPlaygroundTasks {
+                loader_task: VmPlaygroundLoaderTask {
+                    inner: loader_task_receiver,
+                },
+                output_handler_factory_task,
+            },
+        ))
+    }
+
+    #[cfg(test)]
+    pub(crate) fn io(&self) -> &VmPlaygroundIo {
+        &self.io
+    }
+
+    #[tracing::instrument(skip(self), err)]
+    async fn reset_rocksdb_cache(&self, last_retained_batch: L1BatchNumber) -> anyhow::Result<()> {
+        let builder = RocksdbStorage::builder(self.rocksdb_path.as_ref()).await?;
+        let current_l1_batch = builder.l1_batch_number().await;
+        if current_l1_batch <= Some(last_retained_batch) {
+            tracing::info!("Resetting RocksDB cache is not required: its current batch #{current_l1_batch:?} is lower than the target");
+            return Ok(());
+        }
+
+        tracing::info!("Resetting RocksDB cache from batch #{current_l1_batch:?}");
+        let mut conn = self.pool.connection_tagged("vm_playground").await?;
+        builder.roll_back(&mut conn, last_retained_batch).await
+    }
+
+    /// Continuously loads new available batches and writes the corresponding data
+    /// produced by that batch.
+    ///
+    /// # Errors
+    ///
+    /// Propagates RocksDB and Postgres errors.
+    pub async fn run(self, stop_receiver: &watch::Receiver<bool>) -> anyhow::Result<()> {
+        fs::create_dir_all(&self.rocksdb_path)
+            .await
+            .with_context(|| format!("cannot create dir `{}`", self.rocksdb_path))?;
+
+        if let Some(reset_to_batch) = self.reset_to_batch {
+            self.reset_rocksdb_cache(reset_to_batch).await?;
+            self.io
+                .write_cursor(reset_to_batch)
+                .await
+                .context("failed resetting VM playground state")?;
+            tracing::info!("Finished resetting playground state");
+        }
+
+        let (loader, loader_task) = VmRunnerStorage::new(
+            self.pool.clone(),
+            self.rocksdb_path,
+            self.io.clone(),
+            self.chain_id,
+        )
+        .await?;
+        self.loader_task_sender.send(loader_task).ok();
+        let vm_runner = VmRunner::new(
+            self.pool,
+            Box::new(self.io),
+            Arc::new(loader),
+            Box::new(self.output_handler_factory),
+            self.batch_executor,
+        );
+        vm_runner.run(stop_receiver).await
+    }
+}
+
+/// Loader task for the VM playground.
+#[derive(Debug)]
+pub struct VmPlaygroundLoaderTask {
+    inner: oneshot::Receiver<StorageSyncTask<VmPlaygroundIo>>,
+}
+
+impl VmPlaygroundLoaderTask {
+    /// Runs a task until a stop signal is received.
+    pub async fn run(self, mut stop_receiver: watch::Receiver<bool>) -> anyhow::Result<()> {
+        let task = tokio::select! {
+            biased;
+            _ = stop_receiver.changed() => return Ok(()),
+            res = self.inner => match res {
+                Ok(task) => task,
+                Err(_) => anyhow::bail!("VM playground stopped before spawning loader task"),
+            }
+        };
+        task.run(stop_receiver).await
+    }
+}
+
+/// Collection of tasks that need to be run in order for the VM playground to work as intended.
+#[derive(Debug)]
+pub struct VmPlaygroundTasks {
+    /// Task that synchronizes storage with new available batches.
+    pub loader_task: VmPlaygroundLoaderTask,
+    /// Task that handles output from processed batches.
+    pub output_handler_factory_task: ConcurrentOutputHandlerFactoryTask<VmPlaygroundIo>,
+}
+
+/// I/O powering [`VmPlayground`].
+#[derive(Debug, Clone)]
+pub struct VmPlaygroundIo {
+    cursor_file_path: PathBuf,
+    // We don't read this value from the cursor file in the `VmRunnerIo` implementation because reads / writes
+    // aren't guaranteed to be atomic.
+    latest_processed_batch: Arc<watch::Sender<L1BatchNumber>>,
+}
+
+impl VmPlaygroundIo {
+    async fn read_cursor(cursor_file_path: &Path) -> anyhow::Result<Option<L1BatchNumber>> {
+        match fs::read_to_string(cursor_file_path).await {
+            Ok(buffer) => {
+                let cursor = buffer
+                    .parse::<u32>()
+                    .with_context(|| format!("invalid cursor value: {buffer}"))?;
+                Ok(Some(L1BatchNumber(cursor)))
+            }
+            Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(None),
+            Err(err) => Err(anyhow::Error::new(err).context(format!(
+                "failed reading VM playground cursor from `{}`",
+                cursor_file_path.display()
+            ))),
+        }
+    }
+
+    async fn write_cursor(&self, cursor: L1BatchNumber) -> anyhow::Result<()> {
+        let buffer = cursor.to_string();
+        fs::write(&self.cursor_file_path, buffer)
+            .await
+            .with_context(|| {
+                format!(
+                    "failed writing VM playground cursor to `{}`",
+                    self.cursor_file_path.display()
+                )
+            })
+    }
+
+    #[cfg(test)]
+    pub(crate) fn subscribe_to_completed_batches(&self) -> watch::Receiver<L1BatchNumber> {
+        self.latest_processed_batch.subscribe()
+    }
+}
+
+#[async_trait]
+impl VmRunnerIo for VmPlaygroundIo {
+    fn name(&self) -> &'static str {
+        "vm_playground"
+    }
+
+    async fn latest_processed_batch(
+        &self,
+        _conn: &mut Connection<'_, Core>,
+    ) -> anyhow::Result<L1BatchNumber> {
+        Ok(*self.latest_processed_batch.borrow())
+    }
+
+    async fn last_ready_to_be_loaded_batch(
+        &self,
+        conn: &mut Connection<'_, Core>,
+    ) -> anyhow::Result<L1BatchNumber> {
+        let sealed_l1_batch = conn
+            .blocks_dal()
+            .get_sealed_l1_batch_number()
+            .await?
+            .context("no L1 batches in Postgres")?;
+        let last_processed_l1_batch = self.latest_processed_batch(conn).await?;
+        Ok(sealed_l1_batch.min(last_processed_l1_batch + 1))
+    }
+
+    async fn mark_l1_batch_as_processing(
+        &self,
+        _conn: &mut Connection<'_, Core>,
+        l1_batch_number: L1BatchNumber,
+    ) -> anyhow::Result<()> {
+        tracing::info!("Started processing L1 batch #{l1_batch_number}");
+        Ok(())
+    }
+
+    async fn mark_l1_batch_as_completed(
+        &self,
+        _conn: &mut Connection<'_, Core>,
+        l1_batch_number: L1BatchNumber,
+    ) -> anyhow::Result<()> {
+        tracing::info!("Finished processing L1 batch #{l1_batch_number}");
+        self.write_cursor(l1_batch_number).await?;
+        // We should only update the in-memory value after the write to the cursor file succeeded.
+        self.latest_processed_batch.send_replace(l1_batch_number);
+        Ok(())
+    }
+}
+
+#[derive(Debug)]
+struct VmPlaygroundOutputHandler;
+
+#[async_trait]
+impl StateKeeperOutputHandler for VmPlaygroundOutputHandler {
+    async fn handle_l2_block(&mut self, updates_manager: &UpdatesManager) -> anyhow::Result<()> {
+        tracing::trace!("Processed L2 block #{}", updates_manager.l2_block.number);
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl OutputHandlerFactory for VmPlaygroundOutputHandler {
+    async fn create_handler(
+        &mut self,
+        _l1_batch_number: L1BatchNumber,
+    ) -> anyhow::Result<Box<dyn StateKeeperOutputHandler>> {
+        Ok(Box::new(Self))
+    }
+}
diff --git a/core/node/vm_runner/src/lib.rs b/core/node/vm_runner/src/lib.rs
index b252eebcbb1f..03e3f43baedc 100644
--- a/core/node/vm_runner/src/lib.rs
+++ b/core/node/vm_runner/src/lib.rs
@@ -3,7 +3,7 @@
 
 #![warn(missing_debug_implementations, missing_docs)]
 
-mod impls;
+pub mod impls;
 mod io;
 mod output_handler;
 mod process;
@@ -13,13 +13,11 @@ mod metrics;
 #[cfg(test)]
 mod tests;
 
-pub use impls::{
-    BasicWitnessInputProducer, BasicWitnessInputProducerIo, BasicWitnessInputProducerTasks,
-    ProtectiveReadsIo, ProtectiveReadsWriter, ProtectiveReadsWriterTasks,
+pub use self::{
+    io::VmRunnerIo,
+    output_handler::{
+        ConcurrentOutputHandlerFactory, ConcurrentOutputHandlerFactoryTask, OutputHandlerFactory,
+    },
+    process::VmRunner,
+    storage::{BatchExecuteData, StorageSyncTask, VmRunnerStorage},
 };
-pub use io::VmRunnerIo;
-pub use output_handler::{
-    ConcurrentOutputHandlerFactory, ConcurrentOutputHandlerFactoryTask, OutputHandlerFactory,
-};
-pub use process::VmRunner;
-pub use storage::{BatchExecuteData, StorageSyncTask, VmRunnerStorage};
diff --git a/core/node/vm_runner/src/tests/mod.rs b/core/node/vm_runner/src/tests/mod.rs
index 50acba610ba5..c1f1510442ba 100644
--- a/core/node/vm_runner/src/tests/mod.rs
+++ b/core/node/vm_runner/src/tests/mod.rs
@@ -25,6 +25,7 @@ use zksync_utils::u256_to_h256;
 use super::{OutputHandlerFactory, VmRunnerIo};
 
 mod output_handler;
+mod playground;
 mod process;
 mod storage;
 
@@ -271,11 +272,12 @@ async fn store_l1_batches(
         digest.push_tx_hash(tx.hash());
         new_l2_block.hash = digest.finalize(ProtocolVersionId::latest());
 
-        l2_block_number += 1;
         new_l2_block.base_system_contracts_hashes = contract_hashes;
         new_l2_block.l2_tx_count = 1;
         conn.blocks_dal().insert_l2_block(&new_l2_block).await?;
         last_l2_block_hash = new_l2_block.hash;
+        l2_block_number += 1;
+
         let tx_result = execute_l2_transaction(tx.clone());
         conn.transactions_dal()
             .mark_txs_as_executed_in_l2_block(
@@ -289,16 +291,15 @@ async fn store_l1_batches(
 
         // Insert a fictive L2 block at the end of the batch
         let mut fictive_l2_block = create_l2_block(l2_block_number.0);
-        let mut digest = L2BlockHasher::new(
+        let digest = L2BlockHasher::new(
             fictive_l2_block.number,
             fictive_l2_block.timestamp,
             last_l2_block_hash,
         );
-        digest.push_tx_hash(tx.hash());
         fictive_l2_block.hash = digest.finalize(ProtocolVersionId::latest());
-        l2_block_number += 1;
         conn.blocks_dal().insert_l2_block(&fictive_l2_block).await?;
         last_l2_block_hash = fictive_l2_block.hash;
+        l2_block_number += 1;
 
         let header = L1BatchHeader::new(
             l1_batch_number,
@@ -337,9 +338,7 @@ async fn store_l1_batches(
     Ok(batches)
 }
 
-async fn fund(pool: &ConnectionPool<Core>, accounts: &[Account]) {
-    let mut conn = pool.connection().await.unwrap();
-
+async fn fund(conn: &mut Connection<'_, Core>, accounts: &[Account]) {
     let eth_amount = U256::from(10).pow(U256::from(32)); //10^32 wei
 
     for account in accounts {
diff --git a/core/node/vm_runner/src/tests/playground.rs b/core/node/vm_runner/src/tests/playground.rs
new file mode 100644
index 000000000000..772917f00d67
--- /dev/null
+++ b/core/node/vm_runner/src/tests/playground.rs
@@ -0,0 +1,130 @@
+use test_casing::test_casing;
+use tokio::sync::watch;
+use zksync_node_genesis::{insert_genesis_batch, GenesisParams};
+use zksync_state::RocksdbStorage;
+use zksync_state_keeper::MainBatchExecutor;
+use zksync_types::vm::FastVmMode;
+
+use super::*;
+use crate::impls::VmPlayground;
+
+async fn run_playground(
+    pool: ConnectionPool<Core>,
+    rocksdb_dir: &tempfile::TempDir,
+    reset_state: bool,
+) {
+    let mut conn = pool.connection().await.unwrap();
+    let genesis_params = GenesisParams::mock();
+    if conn.blocks_dal().is_genesis_needed().await.unwrap() {
+        insert_genesis_batch(&mut conn, &genesis_params)
+            .await
+            .unwrap();
+
+        // Generate some batches and persist them in Postgres
+        let mut accounts = [Account::random()];
+        fund(&mut conn, &accounts).await;
+        store_l1_batches(
+            &mut conn,
+            1..=1, // TODO: test on >1 batch
+            genesis_params.base_system_contracts().hashes(),
+            &mut accounts,
+        )
+        .await
+        .unwrap();
+    }
+
+    let mut batch_executor = MainBatchExecutor::new(false, false);
+    batch_executor.set_fast_vm_mode(FastVmMode::Shadow);
+    let (playground, playground_tasks) = VmPlayground::new(
+        pool.clone(),
+        Box::new(batch_executor),
+        rocksdb_dir.path().to_str().unwrap().to_owned(),
+        genesis_params.config().l2_chain_id,
+        L1BatchNumber(0),
+        reset_state,
+    )
+    .await
+    .unwrap();
+
+    let (stop_sender, stop_receiver) = watch::channel(false);
+    let playground_io = playground.io().clone();
+    assert_eq!(
+        playground_io
+            .latest_processed_batch(&mut conn)
+            .await
+            .unwrap(),
+        L1BatchNumber(0)
+    );
+    assert_eq!(
+        playground_io
+            .last_ready_to_be_loaded_batch(&mut conn)
+            .await
+            .unwrap(),
+        L1BatchNumber(1)
+    );
+
+    let mut completed_batches = playground_io.subscribe_to_completed_batches();
+    let task_handles = [
+        tokio::spawn(playground_tasks.loader_task.run(stop_receiver.clone())),
+        tokio::spawn(
+            playground_tasks
+                .output_handler_factory_task
+                .run(stop_receiver.clone()),
+        ),
+        tokio::spawn(async move { playground.run(&stop_receiver).await }),
+    ];
+    // Wait until all batches are processed.
+    completed_batches
+        .wait_for(|&number| number == L1BatchNumber(1))
+        .await
+        .unwrap();
+
+    // Check that playground I/O works correctly.
+    assert_eq!(
+        playground_io
+            .latest_processed_batch(&mut conn)
+            .await
+            .unwrap(),
+        L1BatchNumber(1)
+    );
+    // There's no batch #2 in storage
+    assert_eq!(
+        playground_io
+            .last_ready_to_be_loaded_batch(&mut conn)
+            .await
+            .unwrap(),
+        L1BatchNumber(1)
+    );
+
+    stop_sender.send_replace(true);
+    for task_handle in task_handles {
+        task_handle.await.unwrap().unwrap();
+    }
+}
+
+#[test_casing(2, [false, true])]
+#[tokio::test]
+async fn vm_playground_basics(reset_state: bool) {
+    let pool = ConnectionPool::test_pool().await;
+    let rocksdb_dir = tempfile::TempDir::new().unwrap();
+    run_playground(pool, &rocksdb_dir, reset_state).await;
+}
+
+#[tokio::test]
+async fn resetting_playground_state() {
+    let pool = ConnectionPool::test_pool().await;
+    let rocksdb_dir = tempfile::TempDir::new().unwrap();
+    run_playground(pool.clone(), &rocksdb_dir, false).await;
+
+    // Manually catch up RocksDB to Postgres to ensure that resetting it is not trivial.
+    let (_stop_sender, stop_receiver) = watch::channel(false);
+    let mut conn = pool.connection().await.unwrap();
+    RocksdbStorage::builder(rocksdb_dir.path())
+        .await
+        .unwrap()
+        .synchronize(&mut conn, &stop_receiver, None)
+        .await
+        .unwrap();
+
+    run_playground(pool.clone(), &rocksdb_dir, true).await;
+}
diff --git a/core/node/vm_runner/src/tests/process.rs b/core/node/vm_runner/src/tests/process.rs
index 664bdeebf855..ebd7894db6e8 100644
--- a/core/node/vm_runner/src/tests/process.rs
+++ b/core/node/vm_runner/src/tests/process.rs
@@ -31,7 +31,7 @@ async fn process_one_batch() -> anyhow::Result<()> {
     let alice = Account::random();
     let bob = Account::random();
     let mut accounts = vec![alice, bob];
-    fund(&connection_pool, &accounts).await;
+    fund(&mut conn, &accounts).await;
 
     let batches = store_l1_batches(
         &mut conn,
diff --git a/core/node/vm_runner/src/tests/storage.rs b/core/node/vm_runner/src/tests/storage.rs
index 52de43801ff0..3df6afedabcb 100644
--- a/core/node/vm_runner/src/tests/storage.rs
+++ b/core/node/vm_runner/src/tests/storage.rs
@@ -121,11 +121,11 @@ async fn rerun_storage_on_existing_data() -> anyhow::Result<()> {
     insert_genesis_batch(&mut conn, &genesis_params)
         .await
         .unwrap();
-    drop(conn);
     let alice = Account::random();
     let bob = Account::random();
     let mut accounts = vec![alice, bob];
-    fund(&connection_pool, &accounts).await;
+    fund(&mut conn, &accounts).await;
+    drop(conn);
 
     // Generate 10 batches worth of data and persist it in Postgres
     let batches = store_l1_batches(
@@ -212,11 +212,11 @@ async fn continuously_load_new_batches() -> anyhow::Result<()> {
     insert_genesis_batch(&mut conn, &genesis_params)
         .await
         .unwrap();
-    drop(conn);
     let alice = Account::random();
     let bob = Account::random();
     let mut accounts = vec![alice, bob];
-    fund(&connection_pool, &accounts).await;
+    fund(&mut conn, &accounts).await;
+    drop(conn);
 
     let mut tester = StorageTester::new(connection_pool.clone());
     let io_mock = Arc::new(RwLock::new(IoMock::default()));
@@ -284,11 +284,11 @@ async fn access_vm_runner_storage() -> anyhow::Result<()> {
     insert_genesis_batch(&mut conn, &genesis_params)
         .await
         .unwrap();
-    drop(conn);
     let alice = Account::random();
     let bob = Account::random();
     let mut accounts = vec![alice, bob];
-    fund(&connection_pool, &accounts).await;
+    fund(&mut conn, &accounts).await;
+    drop(conn);
 
     // Generate 10 batches worth of data and persist it in Postgres
     let batch_range = 1..=10;