pub async fn new(
        ctx: &ctx::Ctx,
        pool: ConnectionPool,
    ) -> ctx::Result<(Self, StateKeeperRunner)> {
        let mut conn = pool.connection(ctx).await.wrap("connection()")?;
        // We fetch the last protocol version from storage.
        // `protocol_version_id_by_timestamp` does a wrapping conversion to `i64`.
        let protocol_version = ctx
            .wait(
                conn.0
                    .protocol_versions_dal()
                    .protocol_version_id_by_timestamp(i64::MAX.try_into().unwrap()),
            )
            .await?
            .context("protocol_version_id_by_timestamp()")?;
        let cursor = ctx
            .wait(IoCursor::for_fetcher(&mut conn.0))
            .await?
            .context("IoCursor::new()")?;
        let pending_batch = ctx
            .wait(conn.0.blocks_dal().pending_batch_exists())
            .await?
            .context("pending_batch_exists()")?;
        let (actions_sender, actions_queue) = ActionQueue::new();
        let addr = sync::watch::channel(None).0;
        let sync_state = SyncState::default();

        let rocksdb_dir = tempfile::tempdir().context("tempdir()")?;
        let merkle_tree_config = MerkleTreeConfig {
            mode: MerkleTreeMode::Lightweight,
            ..MerkleTreeConfig::for_tests(rocksdb_dir.path().join("merkle_tree"))
        };
        let operation_manager_config = OperationsManagerConfig {
            delay_interval: Duration::from_millis(100),
        };
        let state_keeper_config = StateKeeperConfig {
            protective_reads_persistence_enabled: true,
            ..StateKeeperConfig::for_tests()
        };
        let config = MetadataCalculatorConfig::for_main_node(
            &merkle_tree_config,
            &operation_manager_config,
            &state_keeper_config,
        );
        let metadata_calculator = MetadataCalculator::new(config, None, pool.0.clone())
            .await
            .context("MetadataCalculator::new()")?;
        Ok((
            Self {
                protocol_version,
                last_batch: cursor.l1_batch,
                last_block: cursor.next_l2_block - 1,
                last_timestamp: cursor.prev_l2_block_timestamp,
                batch_sealed: !pending_batch,
                next_priority_op: PriorityOpId(1),
                actions_sender,
                sync_state: sync_state.clone(),
                addr: addr.subscribe(),
                pool: pool.clone(),
            },
            StateKeeperRunner {
                actions_queue,
                sync_state,
                pool: pool.clone(),
                addr,
                rocksdb_dir,
                metadata_calculator,
            },
        ))
    }
Constructs and initializes a new `StateKeeper`.
Caller has to run `StateKeeperRunner.run()` task in the background.
core/node/consensus/src/testonly.rs