pub async fn ensure_ready(
        self,
        config: &MetadataCalculatorRecoveryConfig,
        main_pool: &ConnectionPool<Core>,
        recovery_pool: ConnectionPool<Core>,
        health_updater: &HealthUpdater,
        stop_receiver: &watch::Receiver<bool>,
    ) -> Result<AsyncTree, OrStopped> {
        let started_at = Instant::now();
        let (tree, init_params) = match self {
            Self::Ready(tree) => return Ok(tree),
            Self::Recovering(tree) => {
                let params = InitParameters::new(main_pool, config).await?.context(
                    "Merkle tree is recovering, but Postgres doesn't contain snapshot recovery information",
                )?;

                let recovered_version = tree.recovered_version();
                if u64::from(params.l1_batch.0) != recovered_version {
                    let err = anyhow::anyhow!(
                        "Snapshot L1 batch in Postgres ({params:?}) differs from the recovered Merkle tree version \
                         ({recovered_version})"
                    );
                    return Err(err.into());
                }

                tracing::info!("Resuming tree recovery with status: {params:?}");
                (tree, params)
            }
            Self::Empty { db, mode } => {
                if let Some(params) = InitParameters::new(main_pool, config).await? {
                    tracing::info!("Starting Merkle tree recovery with status {params:?}");
                    let l1_batch = params.l1_batch;
                    let tree = AsyncTreeRecovery::new(db, l1_batch.0.into(), mode, config)?;
                    (tree, params)
                } else {
                    // The genesis block will be filled in `TreeUpdater::loop_updating_tree()`.
                    tracing::info!("Starting Merkle tree from scratch");
                    return Ok(AsyncTree::new(db, mode)?);
                }
            }
        };

        tracing::debug!(
            "Obtained recovery init parameters: {init_params:?} based on recovery configuration {config:?}"
        );
        let recovery_options = RecoveryOptions {
            chunk_count: init_params.chunk_count(),
            concurrency_limit: recovery_pool.max_size() as usize,
            events: Box::new(RecoveryHealthUpdater::new(health_updater)),
        };
        let tree = tree
            .recover(init_params, recovery_options, &recovery_pool, stop_receiver)
            .await?;
        // Only report latency if recovery wasn't canceled
        let elapsed = started_at.elapsed();
        APP_METRICS.snapshot_recovery_latency[&SnapshotRecoveryStage::Tree].set(elapsed);
        tracing::info!("Recovered Merkle tree from snapshot in {elapsed:?}");
        Ok(tree)
    }
Ensures that the tree is ready for the normal operation, recovering it from a Postgres snapshot
if necessary.

`recovery_pool` is taken by value to free up its connection after recovery (provided that it's not shared
with other components).
core/node/metadata_calculator/src/recovery/mod.rs