pub async fn run(self, stop_receiver: watch::Receiver<bool>) -> anyhow::Result<()> {
        if let VmPlaygroundStorage::Rocksdb { path, .. } = &self.storage {
            fs::create_dir_all(path)
                .await
                .with_context(|| format!("cannot create dir `{path:?}`"))?;
        }

        if let Some(reset_to_batch) = self.reset_to_batch {
            self.io.health_updater.update(HealthStatus::Affected.into());

            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");
        }

        self.io.update_health();

        let loader: Arc<dyn StorageLoader> = match self.storage {
            VmPlaygroundStorage::Rocksdb { path, task_sender } => {
                let (loader, loader_task) =
                    VmRunnerStorage::new(self.pool.clone(), path, self.io.clone(), self.chain_id)
                        .await?;
                task_sender.send(loader_task).ok();
                Arc::new(loader)
            }
            VmPlaygroundStorage::Snapshots { shadow } => {
                let mut loader = PostgresLoader::new(self.pool.clone(), self.chain_id).await?;
                loader.shadow_snapshots(shadow);
                Arc::new(loader)
            }
        };
        let vm_runner = VmRunner::new(
            self.pool,
            Arc::new(self.io),
            loader,
            Arc::new(self.output_handler_factory),
            Box::new(self.batch_executor_factory),
        );
        vm_runner.run(&stop_receiver).await
    }
Continuously loads new available batches and writes the corresponding data
produced by that batch.

# Errors

Propagates RocksDB and Postgres errors.
core/node/vm_runner/src/impls/playground.rs