async fn recover_from_snapshot(
        &mut self,
        pool: &ConnectionPool<Core>,
        init_parameters: &InitParameters,
        stop_receiver: &watch::Receiver<bool>,
    ) -> Result<(), OrStopped> {
        if *stop_receiver.borrow() {
            return Err(OrStopped::Stopped);
        }

        tracing::info!(
            ?init_parameters,
            concurrency = pool.max_size(),
            "Recovering secondary storage from snapshot"
        );

        let mut storage = pool.connection_tagged("state_keeper").await?;
        self.set_l1_batch_number(init_parameters.l1_batch, true)
            .await?;
        self.recover_factory_deps(&mut storage, init_parameters)
            .await?;

        if *stop_receiver.borrow() {
            return Err(OrStopped::Stopped);
        }
        let key_chunks = Self::load_key_chunks(&mut storage, init_parameters).await?;
        drop(storage);

        let semaphore = Semaphore::new(pool.max_size() as usize);
        let total_chunk_count = key_chunks.len();
        let key_chunks = self.filter_key_chunks(key_chunks, init_parameters).await?;
        let recovered_chunk_count = total_chunk_count - key_chunks.len();
        RECOVERY_METRICS
            .recovered_chunk_count
            .set(recovered_chunk_count);
        tracing::info!(
            "Checked chunk starts; {recovered_chunk_count} chunks are already recovered"
        );

        let db_mutex = Mutex::new(self.clone());
        let chunk_tasks = key_chunks.into_iter().map(|chunk| async {
            let _permit = semaphore
                .acquire()
                .await
                .context("semaphore is never closed")?;
            Self::recover_logs_chunk(
                &db_mutex,
                pool,
                init_parameters,
                chunk,
                total_chunk_count,
                stop_receiver,
            )
            .await
        });
        future::try_join_all(chunk_tasks).await?;
        drop(db_mutex);

        tracing::info!("All chunks recovered; finalizing recovery process");
        self.save(Some(init_parameters.l1_batch + 1)).await?;
        Ok(())
    }
# Important

`Self::L1_BATCH_NUMBER_KEY` must be set at the very end of the process. If it is set earlier, recovery is not fault-tolerant
(it would be considered complete even if it failed in the middle).
core/lib/state/src/rocksdb/recovery.rs