fix(en): Fix miscellaneous snapshot recovery nits

Cargo.toml

@@ -85,6 +85,7 @@ clap = "4.2.2"
 codegen = "0.2.0"
 criterion = "0.4.0"
 ctrlc = "3.1"
+derive_more = "=1.0.0-beta.6"
 envy = "0.4"
 ethabi = "18.0.0"
 flate2 = "1.0.28"

core/bin/external_node/src/config/mod.rs

@@ -26,6 +26,7 @@ use zksync_core::{
 #[cfg(test)]
 use zksync_dal::{ConnectionPool, Core};
 use zksync_protobuf_config::proto;
+use zksync_snapshots_applier::SnapshotsApplierConfig;
 use zksync_types::{api::BridgeAddresses, fee_model::FeeParams};
 use zksync_web3_decl::{
     client::L2Client,
@@ -320,6 +321,11 @@ pub(crate) struct OptionalENConfig {
     /// This is an experimental and incomplete feature; do not use unless you know what you're doing.
     #[serde(default)]
     pub snapshots_recovery_enabled: bool,
+    /// Maximum concurrency factor for the concurrent parts of snapshot recovery for Postgres. It may be useful to
+    /// reduce this factor to about 5 if snapshot recovery overloads I/O capacity of the node. Conversely,
+    /// if I/O capacity of your infra is high, you may increase concurrency to speed up Postgres recovery.
+    #[serde(default = "OptionalENConfig::default_snapshots_recovery_postgres_max_concurrency")]
+    pub snapshots_recovery_postgres_max_concurrency: NonZeroUsize,
 
     #[serde(default = "OptionalENConfig::default_pruning_chunk_size")]
     pub pruning_chunk_size: u32,
@@ -461,6 +467,10 @@ impl OptionalENConfig {
         L1BatchCommitDataGeneratorMode::Rollup
     }
 
+    fn default_snapshots_recovery_postgres_max_concurrency() -> NonZeroUsize {
+        SnapshotsApplierConfig::default().max_concurrency
+    }
+
     const fn default_pruning_chunk_size() -> u32 {
         10
     }

core/bin/external_node/src/init.rs

@@ -6,7 +6,7 @@ use zksync_core::sync_layer::genesis::perform_genesis_if_needed;
 use zksync_dal::{ConnectionPool, Core, CoreDal};
 use zksync_health_check::AppHealthCheck;
 use zksync_object_store::ObjectStoreFactory;
-use zksync_snapshots_applier::SnapshotsApplierConfig;
+use zksync_snapshots_applier::{SnapshotsApplierConfig, SnapshotsApplierTask};
 use zksync_web3_decl::client::BoxedL2Client;
 
 use crate::config::read_snapshots_recovery_config;
@@ -20,7 +20,7 @@ enum InitDecision {
 }
 
 pub(crate) async fn ensure_storage_initialized(
-    pool: &ConnectionPool<Core>,
+    pool: ConnectionPool<Core>,
     main_node_client: BoxedL2Client,
     app_health: &AppHealthCheck,
     l2_chain_id: L2ChainId,
@@ -89,13 +89,15 @@ pub(crate) async fn ensure_storage_initialized(
                 .await;
 
             let config = SnapshotsApplierConfig::default();
-            app_health.insert_component(config.health_check());
-            config
-                .run(
-                    pool,
-                    &main_node_client.for_component("snapshot_recovery"),
-                    &blob_store,
-                )
+            let snapshots_applier_task = SnapshotsApplierTask::new(
+                config,
+                pool,
+                Box::new(main_node_client.for_component("snapshot_recovery")),
+                blob_store,
+            );
+            app_health.insert_component(snapshots_applier_task.health_check());
+            snapshots_applier_task
+                .run()
                 .await
                 .context("snapshot recovery failed")?;
             tracing::info!("Snapshot recovery is complete");

core/bin/external_node/src/main.rs

@@ -909,7 +909,7 @@ async fn run_node(
 
     // Make sure that the node storage is initialized either via genesis or snapshot recovery.
     ensure_storage_initialized(
-        &connection_pool,
+        connection_pool.clone(),
         main_node_client.clone(),
         &app_health,
         config.remote.l2_chain_id,

core/lib/snapshots_applier/Cargo.toml

@@ -28,4 +28,5 @@ thiserror.workspace = true
 serde = { workspace = true, features = ["derive"] }
 
 [dev-dependencies]
+assert_matches.workspace = true
 test-casing.workspace = true