feat(node_framework): Support StateKeeper in the framework

core/bin/external_node/src/main.rs

@@ -101,7 +101,7 @@ async fn build_state_keeper(
         stop_receiver,
         Box::new(io),
         batch_executor_base,
-        Box::new(NoopSealer),
+        Arc::new(NoopSealer),
     ))
 }
 

core/lib/zksync_core/src/consensus/testonly.rs

@@ -1,6 +1,5 @@
 //! Utilities for testing the consensus module.
-
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};
 
 use anyhow::Context as _;
 use rand::{
@@ -445,7 +444,7 @@ impl StateKeeperRunner {
                     stop_receiver,
                     Box::new(io),
                     Box::new(MockBatchExecutorBuilder),
-                    Box::new(NoopSealer),
+                    Arc::new(NoopSealer),
                 )
                 .run(),
             );

core/lib/zksync_core/src/fee_model.rs

@@ -52,7 +52,7 @@ pub trait BatchFeeModelInputProvider: fmt::Debug + 'static + Send + Sync {
 /// it explicitly gets the L1 gas price from the provider and uses it to calculate the batch fee input instead of getting
 /// it from other node.
 #[derive(Debug)]
-pub(crate) struct MainNodeFeeInputProvider {
+pub struct MainNodeFeeInputProvider {
     provider: Arc<GasAdjuster>,
     config: FeeModelConfig,
 }
@@ -74,7 +74,7 @@ impl BatchFeeModelInputProvider for MainNodeFeeInputProvider {
 }
 
 impl MainNodeFeeInputProvider {
-    pub(crate) fn new(provider: Arc<GasAdjuster>, config: FeeModelConfig) -> Self {
+    pub fn new(provider: Arc<GasAdjuster>, config: FeeModelConfig) -> Self {
         Self { provider, config }
     }
 }

core/lib/zksync_core/src/lib.rs

@@ -758,15 +758,15 @@ async fn add_state_keeper_to_task_futures(
         .build()
         .await
         .context("failed to build state_keeper_pool")?;
-    let next_priority_id = state_keeper_pool
-        .access_storage()
-        .await
-        .unwrap()
-        .transactions_dal()
-        .next_priority_id()
-        .await;
-    let mempool = MempoolGuard::new(next_priority_id, mempool_config.capacity);
-    mempool.register_metrics();
+    let mempool = {
+        let mut storage = state_keeper_pool
+            .access_storage()
+            .await
+            .context("Access storage to build mempool")?;
+        let mempool = MempoolGuard::from_storage(&mut storage, mempool_config.capacity).await;
+        mempool.register_metrics();
+        mempool
+    };
 
     let miniblock_sealer_pool = pool_builder
         .build()
@@ -802,9 +802,13 @@ async fn add_state_keeper_to_task_futures(
         .build()
         .await
         .context("failed to build mempool_fetcher_pool")?;
-    let mempool_fetcher = MempoolFetcher::new(mempool, batch_fee_input_provider, mempool_config);
-    let mempool_fetcher_handle =
-        tokio::spawn(mempool_fetcher.run(mempool_fetcher_pool, stop_receiver));
+    let mempool_fetcher = MempoolFetcher::new(
+        mempool,
+        batch_fee_input_provider,
+        mempool_config,
+        mempool_fetcher_pool,
+    );
+    let mempool_fetcher_handle = tokio::spawn(mempool_fetcher.run(stop_receiver));
     task_futures.push(mempool_fetcher_handle);
     Ok(())
 }

core/lib/zksync_core/src/state_keeper/io/mempool.rs

@@ -45,7 +45,7 @@ use crate::{
 /// Decides which batch parameters should be used for the new batch.
 /// This is an IO for the main server application.
 #[derive(Debug)]
-pub(crate) struct MempoolIO {
+pub struct MempoolIO {
     mempool: MempoolGuard,
     pool: ConnectionPool,
     object_store: Arc<dyn ObjectStore>,
@@ -437,7 +437,7 @@ async fn sleep_past(timestamp: u64, miniblock: MiniblockNumber) -> u64 {
 
 impl MempoolIO {
     #[allow(clippy::too_many_arguments)]
-    pub(in crate::state_keeper) async fn new(
+    pub async fn new(
         mempool: MempoolGuard,
         object_store: Arc<dyn ObjectStore>,
         miniblock_sealer_handle: MiniblockSealerHandle,

core/lib/zksync_core/src/state_keeper/io/mod.rs

@@ -13,7 +13,6 @@ use zksync_types::{
     Transaction,
 };
 
-pub(crate) use self::mempool::MempoolIO;
 use super::{
     metrics::{MiniblockQueueStage, MINIBLOCK_METRICS},
     seal_criteria::IoSealCriteria,

core/lib/zksync_core/src/state_keeper/keeper.rs

@@ -1,6 +1,7 @@
 use std::{
     convert::Infallible,
     future::{self, Future},
+    sync::Arc,
     time::{Duration, Instant},
 };
 
@@ -60,15 +61,15 @@ pub struct ZkSyncStateKeeper {
     stop_receiver: watch::Receiver<bool>,
     io: Box<dyn StateKeeperIO>,
     batch_executor_base: Box<dyn L1BatchExecutorBuilder>,
-    sealer: Box<dyn ConditionalSealer>,
+    sealer: Arc<dyn ConditionalSealer>,
 }
 
 impl ZkSyncStateKeeper {
     pub fn new(
         stop_receiver: watch::Receiver<bool>,
         io: Box<dyn StateKeeperIO>,
         batch_executor_base: Box<dyn L1BatchExecutorBuilder>,
-        sealer: Box<dyn ConditionalSealer>,
+        sealer: Arc<dyn ConditionalSealer>,
     ) -> Self {
         Self {
             stop_receiver,

core/lib/zksync_core/src/state_keeper/mempool_actor.rs

@@ -35,6 +35,7 @@ pub async fn l2_tx_filter(
 #[derive(Debug)]
 pub struct MempoolFetcher {
     mempool: MempoolGuard,
+    pool: ConnectionPool,
     batch_fee_input_provider: Arc<dyn BatchFeeModelInputProvider>,
     sync_interval: Duration,
     sync_batch_size: usize,
@@ -48,9 +49,11 @@ impl MempoolFetcher {
         mempool: MempoolGuard,
         batch_fee_input_provider: Arc<dyn BatchFeeModelInputProvider>,
         config: &MempoolConfig,
+        pool: ConnectionPool,
     ) -> Self {
         Self {
             mempool,
+            pool,
             batch_fee_input_provider,
             sync_interval: config.sync_interval(),
             sync_batch_size: config.sync_batch_size,
@@ -60,12 +63,8 @@ impl MempoolFetcher {
         }
     }
 
-    pub async fn run(
-        mut self,
-        pool: ConnectionPool,
-        stop_receiver: watch::Receiver<bool>,
-    ) -> anyhow::Result<()> {
-        let mut storage = pool.access_storage_tagged("state_keeper").await?;
+    pub async fn run(mut self, stop_receiver: watch::Receiver<bool>) -> anyhow::Result<()> {
+        let mut storage = self.pool.access_storage_tagged("state_keeper").await?;
         if let Some(stuck_tx_timeout) = self.stuck_tx_timeout {
             let removed_txs = storage
                 .transactions_dal()
@@ -87,7 +86,7 @@ impl MempoolFetcher {
                 break;
             }
             let latency = KEEPER_METRICS.mempool_sync.start();
-            let mut storage = pool.access_storage_tagged("state_keeper").await?;
+            let mut storage = self.pool.access_storage_tagged("state_keeper").await?;
             let mempool_info = self.mempool.get_mempool_info();
             let protocol_version = pending_protocol_version(&mut storage)
                 .await
@@ -230,12 +229,16 @@ mod tests {
         let (base_fee, gas_per_pubdata) =
             derive_base_fee_and_gas_per_pubdata(fee_input, ProtocolVersionId::latest().into());
 
-        let mut fetcher =
-            MempoolFetcher::new(mempool.clone(), fee_params_provider, &TEST_MEMPOOL_CONFIG);
+        let mut fetcher = MempoolFetcher::new(
+            mempool.clone(),
+            fee_params_provider,
+            &TEST_MEMPOOL_CONFIG,
+            pool.clone(),
+        );
         let (tx_hashes_sender, mut tx_hashes_receiver) = mpsc::unbounded_channel();
         fetcher.transaction_hashes_sender = tx_hashes_sender;
         let (stop_sender, stop_receiver) = watch::channel(false);
-        let fetcher_task = tokio::spawn(fetcher.run(pool.clone(), stop_receiver));
+        let fetcher_task = tokio::spawn(fetcher.run(stop_receiver));
 
         // Add a new transaction to the storage.
         let transaction = create_l2_transaction(base_fee, gas_per_pubdata);
@@ -283,10 +286,14 @@ mod tests {
         let (base_fee, gas_per_pubdata) =
             derive_base_fee_and_gas_per_pubdata(fee_input, ProtocolVersionId::latest().into());
 
-        let fetcher =
-            MempoolFetcher::new(mempool.clone(), fee_params_provider, &TEST_MEMPOOL_CONFIG);
+        let fetcher = MempoolFetcher::new(
+            mempool.clone(),
+            fee_params_provider,
+            &TEST_MEMPOOL_CONFIG,
+            pool.clone(),
+        );
         let (stop_sender, stop_receiver) = watch::channel(false);
-        let fetcher_task = tokio::spawn(fetcher.run(pool.clone(), stop_receiver));
+        let fetcher_task = tokio::spawn(fetcher.run(stop_receiver));
 
         // Add a transaction with insufficient fee to the storage.
         let transaction = create_l2_transaction(base_fee / 2, gas_per_pubdata / 2);
@@ -319,12 +326,16 @@ mod tests {
         let (base_fee, gas_per_pubdata) =
             derive_base_fee_and_gas_per_pubdata(fee_input, ProtocolVersionId::latest().into());
 
-        let mut fetcher =
-            MempoolFetcher::new(mempool.clone(), fee_params_provider, &TEST_MEMPOOL_CONFIG);
+        let mut fetcher = MempoolFetcher::new(
+            mempool.clone(),
+            fee_params_provider,
+            &TEST_MEMPOOL_CONFIG,
+            pool.clone(),
+        );
         let (tx_hashes_sender, mut tx_hashes_receiver) = mpsc::unbounded_channel();
         fetcher.transaction_hashes_sender = tx_hashes_sender;
         let (stop_sender, stop_receiver) = watch::channel(false);
-        let fetcher_task = tokio::spawn(fetcher.run(pool.clone(), stop_receiver));
+        let fetcher_task = tokio::spawn(fetcher.run(stop_receiver));
 
         // Add a new transaction to the storage.
         let transaction = create_l2_transaction(base_fee * 2, gas_per_pubdata * 2);

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

@@ -8,14 +8,13 @@ use zksync_config::{
 use zksync_dal::ConnectionPool;
 use zksync_object_store::ObjectStore;
 
-use self::io::MempoolIO;
 pub use self::{
     batch_executor::{L1BatchExecutorBuilder, MainBatchExecutorBuilder},
-    io::{MiniblockSealer, MiniblockSealerHandle},
+    io::{mempool::MempoolIO, MiniblockSealer, MiniblockSealerHandle, StateKeeperIO},
     keeper::ZkSyncStateKeeper,
-};
-pub(crate) use self::{
-    mempool_actor::MempoolFetcher, seal_criteria::SequencerSealer, types::MempoolGuard,
+    mempool_actor::MempoolFetcher,
+    seal_criteria::SequencerSealer,
+    types::MempoolGuard,
 };
 use crate::fee_model::BatchFeeModelInputProvider;
 
@@ -75,6 +74,6 @@ pub(crate) async fn create_state_keeper(
         stop_receiver,
         Box::new(io),
         Box::new(batch_executor_base),
-        Box::new(sealer),
+        Arc::new(sealer),
     )
 }

core/lib/zksync_core/src/state_keeper/seal_criteria/conditional_sealer.rs

@@ -115,7 +115,7 @@ impl ConditionalSealer for SequencerSealer {
 }
 
 impl SequencerSealer {
-    pub(crate) fn new(config: StateKeeperConfig) -> Self {
+    pub fn new(config: StateKeeperConfig) -> Self {
         let sealers = Self::default_sealers();
         Self { config, sealers }
     }

core/lib/zksync_core/src/state_keeper/tests/tester.rs

@@ -2,6 +2,7 @@ use std::{
     collections::{HashMap, HashSet, VecDeque},
     convert::TryInto,
     fmt,
+    sync::Arc,
     time::{Duration, Instant},
 };
 
@@ -200,7 +201,7 @@ impl TestScenario {
             stop_receiver,
             Box::new(io),
             Box::new(batch_executor_base),
-            Box::new(sealer),
+            Arc::new(sealer),
         );
         let sk_thread = tokio::spawn(sk.run());
 

core/lib/zksync_core/src/state_keeper/types.rs

@@ -4,6 +4,7 @@ use std::{
 };
 
 use multivm::interface::VmExecutionResultAndLogs;
+use zksync_dal::StorageProcessor;
 use zksync_mempool::{L2TxFilter, MempoolInfo, MempoolStore};
 use zksync_types::{
     block::BlockGasCount, tx::ExecutionMetrics, Address, Nonce, PriorityOpId, Transaction,
@@ -16,7 +17,15 @@ use crate::gas_tracker::{gas_count_from_metrics, gas_count_from_tx_and_metrics};
 pub struct MempoolGuard(Arc<Mutex<MempoolStore>>);
 
 impl MempoolGuard {
-    pub fn new(next_priority_id: PriorityOpId, capacity: u64) -> Self {
+    pub async fn from_storage(storage_processor: &mut StorageProcessor<'_>, capacity: u64) -> Self {
+        let next_priority_id = storage_processor
+            .transactions_dal()
+            .next_priority_id()
+            .await;
+        Self::new(next_priority_id, capacity)
+    }
+
+    pub(super) fn new(next_priority_id: PriorityOpId, capacity: u64) -> Self {
         let store = MempoolStore::new(next_priority_id, capacity);
         Self(Arc::new(Mutex::new(store)))
     }

core/lib/zksync_core/src/sync_layer/tests.rs

@@ -3,6 +3,7 @@
 use std::{
     collections::{HashMap, VecDeque},
     iter,
+    sync::Arc,
     time::{Duration, Instant},
 };
 
@@ -94,7 +95,7 @@ impl StateKeeperHandles {
             stop_receiver,
             Box::new(io),
             Box::new(batch_executor_base),
-            Box::new(NoopSealer),
+            Arc::new(NoopSealer),
         );
         Self {
             stop_sender,

core/node/node_framework/Cargo.toml

@@ -18,6 +18,7 @@ zksync_config = { path = "../../lib/config" }
 zksync_object_store = { path = "../../lib/object_store" }
 zksync_core = { path = "../../lib/zksync_core" }
 zksync_storage = { path = "../../lib/storage" }
+zksync_eth_client = { path = "../../lib/eth_client" }
 
 tracing = "0.1"
 thiserror = "1"