revert: "refactor: Rename consensus tasks and split storage (BFT-476)"

core/bin/external_node/src/main.rs

@@ -286,7 +286,7 @@ async fn run_core(
             // but we only need to wait for stop signal once, and it will be propagated to all child contexts.
             let ctx = ctx::root();
             scope::run!(&ctx, |ctx, s| async move {
-                s.spawn_bg(consensus::era::run_external_node(
+                s.spawn_bg(consensus::era::run_en(
                     ctx,
                     cfg,
                     pool,

core/node/consensus/src/batch.rs

@@ -14,7 +14,7 @@ use zksync_types::{
 };
 use zksync_utils::{h256_to_u256, u256_to_h256};
 
-use crate::storage::ConnectionPool;
+use crate::ConnectionPool;
 
 /// Commitment to the last block of a batch.
 pub(crate) struct LastBlockCommit {

core/node/consensus/src/en.rs

@@ -9,8 +9,8 @@ use zksync_node_sync::{
 use zksync_types::L2BlockNumber;
 use zksync_web3_decl::client::{DynClient, L2};
 
-use super::{config, storage::Store, ConsensusConfig, ConsensusSecrets};
-use crate::storage::{self, ConnectionPool};
+use super::{config, storage::Store, ConnectionPool, ConsensusConfig, ConsensusSecrets};
+use crate::storage;
 
 /// External node.
 pub(super) struct EN {

core/node/consensus/src/era.rs

@@ -10,7 +10,7 @@ use zksync_dal::Core;
 use zksync_node_sync::{sync_action::ActionQueueSender, SyncState};
 use zksync_web3_decl::client::{DynClient, L2};
 
-use super::{en, mn, storage::ConnectionPool};
+use super::{en, storage::ConnectionPool};
 
 /// Runs the consensus task in the main node mode.
 pub async fn run_main_node(
@@ -22,7 +22,7 @@ pub async fn run_main_node(
     // Consensus is a new component.
     // For now in case of error we just log it and allow the server
     // to continue running.
-    if let Err(err) = mn::run_main_node(ctx, cfg, secrets, ConnectionPool(pool)).await {
+    if let Err(err) = super::run_main_node(ctx, cfg, secrets, ConnectionPool(pool)).await {
         tracing::error!("Consensus actor failed: {err:#}");
     } else {
         tracing::info!("Consensus actor stopped");
@@ -33,7 +33,7 @@ pub async fn run_main_node(
 /// Runs the consensus node for the external node.
 /// If `cfg` is `None`, it will just fetch blocks from the main node
 /// using JSON RPC, without starting the consensus node.
-pub async fn run_external_node(
+pub async fn run_en(
     ctx: &ctx::Ctx,
     cfg: Option<(ConsensusConfig, ConsensusSecrets)>,
     pool: zksync_dal::ConnectionPool<Core>,

core/node/consensus/src/lib.rs

@@ -2,8 +2,14 @@
 
 #![allow(clippy::redundant_locals)]
 #![allow(clippy::needless_pass_by_ref_mut)]
-
+use anyhow::Context as _;
+use zksync_concurrency::{ctx, error::Wrap as _, scope};
 use zksync_config::configs::consensus::{ConsensusConfig, ConsensusSecrets};
+use zksync_consensus_executor as executor;
+use zksync_consensus_roles::validator;
+use zksync_consensus_storage::{BatchStore, BlockStore};
+
+use crate::storage::{ConnectionPool, Store};
 
 // Currently `batch` module is only used in tests,
 // but will be used in production once batch syncing is implemented in consensus.
@@ -12,9 +18,64 @@ mod batch;
 mod config;
 mod en;
 pub mod era;
-mod mn;
 mod storage;
 #[cfg(test)]
 pub(crate) mod testonly;
 #[cfg(test)]
 mod tests;
+
+/// Task running a consensus validator for the main node.
+/// Main node is currently the only leader of the consensus - i.e. it proposes all the
+/// L2 blocks (generated by `Statekeeper`).
+async fn run_main_node(
+    ctx: &ctx::Ctx,
+    cfg: ConsensusConfig,
+    secrets: ConsensusSecrets,
+    pool: ConnectionPool,
+) -> anyhow::Result<()> {
+    let validator_key = config::validator_key(&secrets)
+        .context("validator_key")?
+        .context("missing validator_key")?;
+    scope::run!(&ctx, |ctx, s| async {
+        if let Some(spec) = &cfg.genesis_spec {
+            let spec = config::GenesisSpec::parse(spec).context("GenesisSpec::parse()")?;
+            pool.connection(ctx)
+                .await
+                .wrap("connection()")?
+                .adjust_genesis(ctx, &spec)
+                .await
+                .wrap("adjust_genesis()")?;
+        }
+        let (store, runner) = Store::new(ctx, pool, None).await.wrap("Store::new()")?;
+        s.spawn_bg(async { runner.run(ctx).await.context("Store::runner()") });
+        let (block_store, runner) = BlockStore::new(ctx, Box::new(store.clone()))
+            .await
+            .wrap("BlockStore::new()")?;
+        s.spawn_bg(async { runner.run(ctx).await.context("BlockStore::runner()") });
+        anyhow::ensure!(
+            block_store.genesis().leader_selection
+                == validator::LeaderSelectionMode::Sticky(validator_key.public()),
+            "unsupported leader selection mode - main node has to be the leader"
+        );
+
+        // Dummy batch store - we don't gossip batches yet, but we need one anyway.
+        let (batch_store, runner) = BatchStore::new(ctx, Box::new(store.clone()))
+            .await
+            .wrap("BatchStore::new()")?;
+        s.spawn_bg(async { runner.run(ctx).await.context("BatchStore::runner()") });
+
+        let executor = executor::Executor {
+            config: config::executor(&cfg, &secrets)?,
+            block_store,
+            batch_store,
+            attester: None,
+            validator: Some(executor::Validator {
+                key: validator_key,
+                replica_store: Box::new(store.clone()),
+                payload_manager: Box::new(store.clone()),
+            }),
+        };
+        executor.run(ctx).await.context("executor.run()")
+    })
+    .await
+}