feat(node-framework): Add consistency checker (#1527)

## What ❔
Consistency-checker component is the component used by EN, so this PR is
the first step of porting EN to the framework.

## Why ❔
step-by-step porting EN to framework

## Checklist

<!-- Check your PR fulfills the following items. -->
<!-- For draft PRs check the boxes as you complete them. -->

- [ ] PR title corresponds to the body of PR (we generate changelog
entries from PRs).
- [ ] Tests for the changes have been added / updated.
- [ ] Documentation comments have been added / updated.
- [ ] Code has been formatted via `zk fmt` and `zk lint`.
- [ ] Spellcheck has been run via `zk spellcheck`.
- [ ] Linkcheck has been run via `zk linkcheck`.

core/bin/external_node/src/main.rs

@@ -291,7 +291,7 @@ async fn init_tasks(
     };
 
     let consistency_checker = ConsistencyChecker::new(
-        Box::new(eth_client),
+        Arc::new(eth_client),
         10, // TODO (BFT-97): Make it a part of a proper EN config
         singleton_pool_builder
             .build()

core/lib/zksync_core/src/consistency_checker/mod.rs

@@ -303,7 +303,7 @@ pub struct ConsistencyChecker {
     /// How many past batches to check when starting
     max_batches_to_recheck: u32,
     sleep_interval: Duration,
-    l1_client: Box<dyn EthInterface>,
+    l1_client: Arc<dyn EthInterface>,
     event_handler: Box<dyn HandleConsistencyCheckerEvent>,
     l1_data_mismatch_behavior: L1DataMismatchBehavior,
     pool: ConnectionPool<Core>,
@@ -315,7 +315,7 @@ impl ConsistencyChecker {
     const DEFAULT_SLEEP_INTERVAL: Duration = Duration::from_secs(5);
 
     pub fn new(
-        l1_client: Box<dyn EthInterface>,
+        l1_client: Arc<dyn EthInterface>,
         max_batches_to_recheck: u32,
         pool: ConnectionPool<Core>,
         l1_batch_commit_data_generator: Arc<dyn L1BatchCommitDataGenerator>,

core/lib/zksync_core/src/consistency_checker/tests/mod.rs

@@ -87,7 +87,7 @@ pub(crate) fn create_mock_checker(
         diamond_proxy_addr: Some(DIAMOND_PROXY_ADDR),
         max_batches_to_recheck: 100,
         sleep_interval: Duration::from_millis(10),
-        l1_client: Box::new(client),
+        l1_client: Arc::new(client),
         event_handler: Box::new(health_updater),
         l1_data_mismatch_behavior: L1DataMismatchBehavior::Bail,
         pool,

core/node/node_framework/src/implementations/layers/consistency_checker.rs

@@ -0,0 +1,85 @@
+use zksync_core::consistency_checker::ConsistencyChecker;
+use zksync_types::Address;
+
+use crate::{
+    implementations::resources::{
+        eth_interface::EthInterfaceResource, healthcheck::AppHealthCheckResource,
+        l1_batch_commit_data_generator::L1BatchCommitDataGeneratorResource,
+        pools::MasterPoolResource,
+    },
+    service::{ServiceContext, StopReceiver},
+    task::Task,
+    wiring_layer::{WiringError, WiringLayer},
+};
+
+#[derive(Debug)]
+pub struct ConsistencyCheckerLayer {
+    diamond_proxy_addr: Address,
+    max_batches_to_recheck: u32,
+}
+
+impl ConsistencyCheckerLayer {
+    pub fn new(
+        diamond_proxy_addr: Address,
+        max_batches_to_recheck: u32,
+    ) -> ConsistencyCheckerLayer {
+        Self {
+            diamond_proxy_addr,
+            max_batches_to_recheck,
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl WiringLayer for ConsistencyCheckerLayer {
+    fn layer_name(&self) -> &'static str {
+        "consistency_checker_layer"
+    }
+
+    async fn wire(self: Box<Self>, mut context: ServiceContext<'_>) -> Result<(), WiringError> {
+        // Get resources.
+        let l1_client = context.get_resource::<EthInterfaceResource>().await?.0;
+
+        let pool_resource = context.get_resource::<MasterPoolResource>().await?;
+        let singleton_pool = pool_resource.get_singleton().await?;
+
+        let l1_batch_commit_data_generator = context
+            .get_resource::<L1BatchCommitDataGeneratorResource>()
+            .await?
+            .0;
+
+        let consistency_checker = ConsistencyChecker::new(
+            l1_client,
+            self.max_batches_to_recheck,
+            singleton_pool,
+            l1_batch_commit_data_generator,
+        )
+        .map_err(WiringError::Internal)?
+        .with_diamond_proxy_addr(self.diamond_proxy_addr);
+
+        let AppHealthCheckResource(app_health) = context.get_resource_or_default().await;
+        app_health.insert_component(consistency_checker.health_check().clone());
+
+        // Create and add tasks.
+        context.add_task(Box::new(ConsistencyCheckerTask {
+            consistency_checker,
+        }));
+
+        Ok(())
+    }
+}
+
+pub struct ConsistencyCheckerTask {
+    consistency_checker: ConsistencyChecker,
+}
+
+#[async_trait::async_trait]
+impl Task for ConsistencyCheckerTask {
+    fn name(&self) -> &'static str {
+        "consistency_checker"
+    }
+
+    async fn run(self: Box<Self>, stop_receiver: StopReceiver) -> anyhow::Result<()> {
+        self.consistency_checker.run(stop_receiver.0).await
+    }
+}

core/node/node_framework/src/implementations/layers/l1_batch_commit_data_generator.rs

@@ -0,0 +1,46 @@
+use std::sync::Arc;
+
+use zksync_config::configs::chain::L1BatchCommitDataGeneratorMode;
+use zksync_core::eth_sender::l1_batch_commit_data_generator::{
+    L1BatchCommitDataGenerator, RollupModeL1BatchCommitDataGenerator,
+    ValidiumModeL1BatchCommitDataGenerator,
+};
+
+use crate::{
+    implementations::resources::l1_batch_commit_data_generator::L1BatchCommitDataGeneratorResource,
+    service::ServiceContext,
+    wiring_layer::{WiringError, WiringLayer},
+};
+
+#[derive(Debug)]
+pub struct L1BatchCommitDataGeneratorLayer {
+    mode: L1BatchCommitDataGeneratorMode,
+}
+
+impl L1BatchCommitDataGeneratorLayer {
+    pub fn new(mode: L1BatchCommitDataGeneratorMode) -> Self {
+        Self { mode }
+    }
+}
+
+#[async_trait::async_trait]
+impl WiringLayer for L1BatchCommitDataGeneratorLayer {
+    fn layer_name(&self) -> &'static str {
+        "l1_batch_commit_data_generator"
+    }
+
+    async fn wire(self: Box<Self>, mut context: ServiceContext<'_>) -> Result<(), WiringError> {
+        let l1_batch_commit_data_generator: Arc<dyn L1BatchCommitDataGenerator> = match self.mode {
+            L1BatchCommitDataGeneratorMode::Rollup => {
+                Arc::new(RollupModeL1BatchCommitDataGenerator {})
+            }
+            L1BatchCommitDataGeneratorMode::Validium => {
+                Arc::new(ValidiumModeL1BatchCommitDataGenerator {})
+            }
+        };
+        context.insert_resource(L1BatchCommitDataGeneratorResource(
+            l1_batch_commit_data_generator,
+        ))?;
+        Ok(())
+    }
+}

core/node/node_framework/src/implementations/layers/mod.rs

@@ -1,10 +1,12 @@
 pub mod circuit_breaker_checker;
 pub mod commitment_generator;
+pub mod consistency_checker;
 pub mod contract_verification_api;
 pub mod eth_sender;
 pub mod eth_watch;
 pub mod healtcheck_server;
 pub mod house_keeper;
+pub mod l1_batch_commit_data_generator;
 pub mod l1_gas;
 pub mod metadata_calculator;
 pub mod object_store;

core/node/node_framework/src/implementations/resources/l1_batch_commit_data_generator.rs

@@ -0,0 +1,14 @@
+use std::sync::Arc;
+
+use zksync_core::eth_sender::l1_batch_commit_data_generator::L1BatchCommitDataGenerator;
+
+use crate::resource::{Resource, ResourceId};
+
+#[derive(Debug, Clone)]
+pub struct L1BatchCommitDataGeneratorResource(pub Arc<dyn L1BatchCommitDataGenerator>);
+
+impl Resource for L1BatchCommitDataGeneratorResource {
+    fn resource_id() -> ResourceId {
+        "common/l1_batch_commit_data_generator".into()
+    }
+}

core/node/node_framework/src/implementations/resources/mod.rs

@@ -2,6 +2,7 @@ pub mod circuit_breakers;
 pub mod eth_interface;
 pub mod fee_input;
 pub mod healthcheck;
+pub mod l1_batch_commit_data_generator;
 pub mod l1_tx_params;
 pub mod object_store;
 pub mod pools;