feat(node-framework): Add reorg detector

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

@@ -16,6 +16,7 @@ pub mod pools_layer;
 pub mod prometheus_exporter;
 pub mod proof_data_handler;
 pub mod query_eth_client;
+pub mod reorg_detector;
 pub mod sigint;
 pub mod state_keeper;
 pub mod web3_api;

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

@@ -0,0 +1,93 @@
+use std::sync::Arc;
+
+use anyhow::Context;
+use zksync_block_reverter::BlockReverter;
+use zksync_core::reorg_detector::{self, ReorgDetector};
+
+use crate::{
+    implementations::resources::{
+        main_node_client::MainNodeClientResource,
+        pools::{MasterPool, PoolResource},
+        reverter::BlockReverterResource,
+    },
+    precondition::Precondition,
+    service::{ServiceContext, StopReceiver},
+    task::Task,
+    wiring_layer::{WiringError, WiringLayer},
+};
+
+#[derive(Debug)]
+pub struct ReorgDetectorLayer;
+
+#[async_trait::async_trait]
+impl WiringLayer for ReorgDetectorLayer {
+    fn layer_name(&self) -> &'static str {
+        "reorg_detector_layer"
+    }
+
+    async fn wire(self: Box<Self>, mut context: ServiceContext<'_>) -> Result<(), WiringError> {
+        // Get resources.
+        let main_node_client = context.get_resource::<MainNodeClientResource>().await?.0;
+
+        let pool_resource = context.get_resource::<PoolResource<MasterPool>>().await?;
+        let pool = pool_resource.get().await?;
+
+        let reverter = context.get_resource::<BlockReverterResource>().await?.0;
+
+        // Create and insert precondition.
+        context.add_precondition(Box::new(ReorgDetectorPrecondition {
+            reorg_detector: ReorgDetector::new(main_node_client.clone(), pool.clone()),
+            reverter,
+        }));
+
+        // Create and insert task.
+        context.add_task(Box::new(ReorgDetectorTask {
+            reorg_detector: ReorgDetector::new(main_node_client, pool),
+        }));
+
+        Ok(())
+    }
+}
+
+pub struct ReorgDetectorPrecondition {
+    reorg_detector: ReorgDetector,
+    reverter: Arc<BlockReverter>,
+}
+
+#[async_trait::async_trait]
+impl Precondition for ReorgDetectorPrecondition {
+    fn name(&self) -> &'static str {
+        "reorg_detector"
+    }
+
+    async fn check(mut self: Box<Self>, _stop_receiver: StopReceiver) -> anyhow::Result<()> {
+        match self.reorg_detector.check_consistency().await {
+            Ok(()) => {}
+            Err(reorg_detector::Error::ReorgDetected(last_correct_l1_batch)) => {
+                tracing::info!("Reverting to l1 batch number {last_correct_l1_batch}");
+                self.reverter.roll_back(last_correct_l1_batch).await?;
+                tracing::info!("Revert successfully completed");
+            }
+            Err(err) => return Err(err).context("reorg_detector.check_consistency()"),
+        }
+        Ok(())
+    }
+}
+
+pub struct ReorgDetectorTask {
+    reorg_detector: ReorgDetector,
+}
+
+#[async_trait::async_trait]
+impl Task for ReorgDetectorTask {
+    fn name(&self) -> &'static str {
+        "reorg_detector"
+    }
+
+    async fn run(mut self: Box<Self>, stop_receiver: StopReceiver) -> anyhow::Result<()> {
+        self.reorg_detector
+            .run(stop_receiver.0)
+            .await
+            .context("reorg_detector.run()")
+    }
+}

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

@@ -8,6 +8,7 @@ pub mod l1_tx_params;
 pub mod main_node_client;
 pub mod object_store;
 pub mod pools;
+pub mod reverter;
 pub mod state_keeper;
 pub mod sync_state;
 pub mod web3_api;

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

@@ -0,0 +1,15 @@
+use std::sync::Arc;
+
+use zksync_block_reverter::BlockReverter;
+
+use crate::resource::Resource;
+
+/// Wrapper for the  block reverter.
+#[derive(Debug, Clone)]
+pub struct BlockReverterResource(pub Arc<BlockReverter>);
+
+impl Resource for BlockReverterResource {
+    fn name() -> String {
+        "common/block_reverter".into()
+    }
+}