Remove core private api

Signed-off-by: deniallugo <[email protected]>

Cargo.toml

@@ -19,6 +19,7 @@ members = [
 
     # Libraries
     "core/lib/circuit",
+    "core/lib/mempool",
     "core/lib/eth_client",
     "core/lib/eth_signer",
     "core/lib/gateway_watcher",

core/bin/server/Cargo.toml

@@ -20,6 +20,8 @@ zksync_forced_exit_requests = { path = "../zksync_forced_exit_requests", version
 
 zksync_prometheus_exporter = { path = "../../lib/prometheus_exporter", version = "1.0" }
 zksync_config = { path = "../../lib/config", version = "1.0" }
+
+zksync_mempool = { path = "../../lib/mempool", version = "1.0" }
 zksync_storage = { path = "../../lib/storage", version = "1.0" }
 zksync_gateway_watcher = { path = "../../lib/gateway_watcher", version = "1.0" }
 zksync_utils = { path = "../../lib/utils", version = "1.0" }

core/bin/server/src/main.rs

@@ -16,17 +16,17 @@ use zksync_witness_generator::run_prover_server;
 use tokio::task::JoinHandle;
 use zksync_config::configs::api::PrometheusConfig;
 use zksync_config::{
-    configs::api::{
-        CommonApiConfig, JsonRpcConfig, PrivateApiConfig, ProverApiConfig, RestApiConfig,
-        Web3Config,
-    },
+    configs::api::{CommonApiConfig, JsonRpcConfig, ProverApiConfig, RestApiConfig, Web3Config},
     ChainConfig, ContractsConfig, DBConfig, ETHClientConfig, ETHSenderConfig, ETHWatchConfig,
     ForcedExitRequestsConfig, GatewayWatcherConfig, ProverConfig, TickerConfig, ZkSyncConfig,
 };
 use zksync_core::rejected_tx_cleaner::run_rejected_tx_cleaner;
+use zksync_mempool::run_mempool_tx_handler;
 use zksync_prometheus_exporter::{run_operation_counter, run_prometheus_exporter};
 use zksync_storage::ConnectionPool;
 
+const DEFAULT_CHANNEL_CAPACITY: usize = 32_768;
+
 #[derive(Debug, Clone, Copy)]
 pub enum ServerCommand {
     Genesis,
@@ -194,7 +194,6 @@ async fn run_server(components: &ComponentsToRun) {
             sign_check_receiver,
         ));
 
-        let private_config = PrivateApiConfig::from_env();
         let contracts_config = ContractsConfig::from_env();
         let common_config = CommonApiConfig::from_env();
         let chain_config = ChainConfig::from_env();
@@ -212,37 +211,59 @@ async fn run_server(components: &ComponentsToRun) {
         );
 
         if components.0.contains(&Component::RpcWebSocketApi) {
+            let (mempool_tx_request_sender, mempool_tx_request_receiver) =
+                mpsc::channel(DEFAULT_CHANNEL_CAPACITY);
+            tasks.push(run_mempool_tx_handler(
+                connection_pool.clone(),
+                mempool_tx_request_receiver,
+                chain_config.state_keeper.block_chunk_sizes.clone(),
+            ));
             tasks.push(zksync_api::api_server::rpc_subscriptions::start_ws_server(
                 connection_pool.clone(),
                 sign_check_sender.clone(),
                 ticker.clone(),
                 &common_config,
                 &JsonRpcConfig::from_env(),
                 chain_config.state_keeper.miniblock_iteration_interval(),
-                private_config.url.clone(),
+                mempool_tx_request_sender,
                 eth_watch_config.confirmations_for_eth_event,
             ));
         }
 
         if components.0.contains(&Component::RpcApi) {
+            let (mempool_tx_request_sender, mempool_tx_request_receiver) =
+                mpsc::channel(DEFAULT_CHANNEL_CAPACITY);
+            tasks.push(run_mempool_tx_handler(
+                connection_pool.clone(),
+                mempool_tx_request_receiver,
+                chain_config.state_keeper.block_chunk_sizes.clone(),
+            ));
             tasks.push(zksync_api::api_server::rpc_server::start_rpc_server(
                 connection_pool.clone(),
                 sign_check_sender.clone(),
                 ticker.clone(),
                 &JsonRpcConfig::from_env(),
                 &common_config,
-                private_config.url,
+                mempool_tx_request_sender,
                 eth_watch_config.confirmations_for_eth_event,
             ));
         }
 
         if components.0.contains(&Component::RestApi) {
+            let (mempool_tx_request_sender, mempool_tx_request_receiver) =
+                mpsc::channel(DEFAULT_CHANNEL_CAPACITY);
+            tasks.push(run_mempool_tx_handler(
+                connection_pool.clone(),
+                mempool_tx_request_receiver,
+                chain_config.state_keeper.block_chunk_sizes.clone(),
+            ));
             zksync_api::api_server::rest::start_server_thread_detached(
                 connection_pool.clone(),
                 RestApiConfig::from_env().bind_addr(),
                 contracts_config.contract_addr,
                 ticker,
                 sign_check_sender,
+                mempool_tx_request_sender,
             );
         }
     }
@@ -279,7 +300,7 @@ async fn run_server(components: &ComponentsToRun) {
     }
 
     if components.0.contains(&Component::ForcedExit) {
-        tasks.push(run_forced_exit(connection_pool.clone()));
+        tasks.append(&mut run_forced_exit(connection_pool.clone()));
     }
 
     if components.0.contains(&Component::RejectedTaskCleaner) {
@@ -306,22 +327,30 @@ async fn run_server(components: &ComponentsToRun) {
     };
 }
 
-pub fn run_forced_exit(connection_pool: ConnectionPool) -> JoinHandle<()> {
+pub fn run_forced_exit(connection_pool: ConnectionPool) -> Vec<JoinHandle<()>> {
     vlog::info!("Starting the ForcedExitRequests actors");
     let config = ForcedExitRequestsConfig::from_env();
     let common_config = CommonApiConfig::from_env();
-    let private_api_config = PrivateApiConfig::from_env();
     let contract_config = ContractsConfig::from_env();
     let eth_client_config = ETHClientConfig::from_env();
-
-    run_forced_exit_requests_actors(
+    let chain_config = ChainConfig::from_env();
+
+    let (mempool_tx_request_sender, mempool_tx_request_receiver) =
+        mpsc::channel(DEFAULT_CHANNEL_CAPACITY);
+    let mempool_task = run_mempool_tx_handler(
+        connection_pool.clone(),
+        mempool_tx_request_receiver,
+        chain_config.state_keeper.block_chunk_sizes.clone(),
+    );
+    let forced_exit_task = run_forced_exit_requests_actors(
         connection_pool,
-        private_api_config.url,
+        mempool_tx_request_sender,
         config,
         common_config,
         contract_config,
         eth_client_config.web3_url(),
-    )
+    );
+    vec![mempool_task, forced_exit_task]
 }
 
 pub fn run_witness_generator(connection_pool: ConnectionPool) -> JoinHandle<()> {

core/bin/zksync_api/Cargo.toml

@@ -19,6 +19,7 @@ zksync_types = { path = "../../lib/types", version = "1.0" }
 zksync_storage = { path = "../../lib/storage", version = "1.0" }
 
 zksync_crypto = { path = "../../lib/crypto", version = "1.0" }
+zksync_mempool = { path = "../../lib/mempool", version = "1.0" }
 zksync_config = { path = "../../lib/config", version = "1.0" }
 zksync_utils = { path = "../../lib/utils", version = "1.0" }
 zksync_contracts = { path = "../../lib/contracts", version = "1.0" }

core/bin/zksync_api/src/api_server/rest/mod.rs

@@ -15,6 +15,7 @@ use super::tx_sender::TxSender;
 use crate::fee_ticker::FeeTicker;
 use tokio::task::JoinHandle;
 use zksync_config::ZkSyncConfig;
+use zksync_mempool::MempoolTransactionRequest;
 
 mod forced_exit_requests;
 mod helpers;
@@ -26,6 +27,7 @@ async fn start_server(
     fee_ticker: FeeTicker,
     sign_verifier: mpsc::Sender<VerifySignatureRequest>,
     bind_to: SocketAddr,
+    mempool_tx_sender: mpsc::Sender<MempoolTransactionRequest>,
 ) {
     HttpServer::new(move || {
         let api_v01 = api_v01.clone();
@@ -47,7 +49,7 @@ async fn start_server(
                 sign_verifier.clone(),
                 fee_ticker.clone(),
                 &api_v01.config.api.common,
-                api_v01.config.api.private.url.clone(),
+                mempool_tx_sender.clone(),
             );
             v02::api_scope(tx_sender, &api_v01.config)
         };
@@ -86,6 +88,7 @@ pub fn start_server_thread_detached(
     contract_address: H160,
     fee_ticker: FeeTicker,
     sign_verifier: mpsc::Sender<VerifySignatureRequest>,
+    mempool_tx_sender: mpsc::Sender<MempoolTransactionRequest>,
 ) -> JoinHandle<()> {
     let (handler, panic_sender) = spawn_panic_handler();
 
@@ -101,7 +104,14 @@ pub fn start_server_thread_detached(
                 let api_v01 = ApiV01::new(connection_pool, contract_address, config);
                 api_v01.spawn_network_status_updater(panic_sender);
 
-                start_server(api_v01, fee_ticker, sign_verifier, listen_addr).await;
+                start_server(
+                    api_v01,
+                    fee_ticker,
+                    sign_verifier,
+                    listen_addr,
+                    mempool_tx_sender.clone(),
+                )
+                .await;
             });
         })
         .expect("Api server thread");

core/bin/zksync_api/src/api_server/rpc_server/mod.rs

@@ -33,6 +33,7 @@ use self::types::*;
 use super::tx_sender::TxSender;
 use crate::fee_ticker::FeeTicker;
 use ip_insert_middleware::IpInsertMiddleWare;
+use zksync_mempool::MempoolTransactionRequest;
 
 #[derive(Clone)]
 pub struct RpcApp {
@@ -51,8 +52,8 @@ impl RpcApp {
         sign_verify_request_sender: mpsc::Sender<VerifySignatureRequest>,
         ticker: FeeTicker,
         config: &CommonApiConfig,
-        private_url: String,
         confirmations_for_eth_event: u64,
+        mempool_tx_sender: mpsc::Sender<MempoolTransactionRequest>,
     ) -> Self {
         let api_requests_caches_size = config.caches_size;
 
@@ -61,7 +62,7 @@ impl RpcApp {
             sign_verify_request_sender,
             ticker,
             config,
-            private_url,
+            mempool_tx_sender,
         );
 
         RpcApp {
@@ -259,7 +260,7 @@ pub fn start_rpc_server(
     ticker: FeeTicker,
     config: &JsonRpcConfig,
     common_api_config: &CommonApiConfig,
-    private_url: String,
+    mempool_tx_sender: mpsc::Sender<MempoolTransactionRequest>,
     confirmations_for_eth_event: u64,
 ) -> JoinHandle<()> {
     let addr = config.http_bind_addr();
@@ -268,8 +269,8 @@ pub fn start_rpc_server(
         sign_verify_request_sender,
         ticker,
         common_api_config,
-        private_url,
         confirmations_for_eth_event,
+        mempool_tx_sender,
     );
 
     let (handler, panic_sender) = spawn_panic_handler();

core/bin/zksync_api/src/api_server/rpc_subscriptions.rs

@@ -12,6 +12,7 @@ use jsonrpc_ws_server::RequestContext;
 use tokio::task::JoinHandle;
 // Workspace uses
 use zksync_config::configs::api::{CommonApiConfig, JsonRpcConfig};
+use zksync_mempool::MempoolTransactionRequest;
 use zksync_storage::ConnectionPool;
 use zksync_types::{tx::TxHash, ActionType, Address};
 use zksync_utils::panic_notify::{spawn_panic_handler, ThreadPanicNotify};
@@ -183,7 +184,7 @@ pub fn start_ws_server(
     common_config: &CommonApiConfig,
     config: &JsonRpcConfig,
     miniblock_iteration_interval: Duration,
-    private_url: String,
+    mempool_tx_sender: mpsc::Sender<MempoolTransactionRequest>,
     confirmations_for_eth_event: u64,
 ) -> JoinHandle<()> {
     let addr = config.ws_bind_addr();
@@ -202,8 +203,8 @@ pub fn start_ws_server(
         sign_verify_request_sender,
         ticker,
         common_config,
-        private_url,
         confirmations_for_eth_event,
+        mempool_tx_sender,
     );
 
     let (handler, panic_sender) = spawn_panic_handler();