refactor: Split the rest of the zksync_core

Cargo.toml

@@ -23,9 +23,18 @@ members = [
     "core/node/eth_sender",
     "core/node/vm_runner",
     "core/node/test_utils",
+    "core/node/state_keeper",
+    "core/node/reorg_detector",
+    "core/node/consistency_checker",
+    "core/node/metadata_calculator",
+    "core/node/node_sync",
+    "core/node/consensus",
+    "core/node/contract_verification_server",
+    "core/node/api_server",
+    "core/node/tee_verifier_input_producer",
     # Libraries
     "core/lib/db_connection",
-    "core/lib/zksync_core",
+    "core/lib/zksync_core_leftovers",
     "core/lib/basic_types",
     "core/lib/config",
     "core/lib/constants",
@@ -202,7 +211,7 @@ zksync_basic_types = { path = "core/lib/basic_types" }
 zksync_circuit_breaker = { path = "core/lib/circuit_breaker" }
 zksync_config = { path = "core/lib/config" }
 zksync_contracts = { path = "core/lib/contracts" }
-zksync_core = { path = "core/lib/zksync_core" }
+zksync_core_leftovers = { path = "core/lib/zksync_core_leftovers" }
 zksync_crypto = { path = "core/lib/crypto" }
 zksync_dal = { path = "core/lib/dal" }
 zksync_db_connection = { path = "core/lib/db_connection" }
@@ -243,3 +252,12 @@ zksync_node_db_pruner = { path = "core/node/db_pruner" }
 zksync_node_fee_model = { path = "core/node/fee_model" }
 zksync_vm_runner = { path = "core/node/vm_runner" }
 zksync_node_test_utils = { path = "core/node/test_utils" }
+zksync_state_keeper = { path = "core/node/state_keeper" }
+zksync_reorg_detector = { path = "core/node/reorg_detector" }
+zksync_consistency_checker = { path = "core/node/consistency_checker" }
+zksync_metadata_calculator = { path = "core/node/metadata_calculator" }
+zksync_node_sync = { path = "core/node/node_sync" }
+zksync_node_consensus = { path = "core/node/consensus" }
+zksync_contract_verification_server = { path = "core/node/contract_verification_server" }
+zksync_node_api_server = { path = "core/node/api_server" }
+zksync_tee_verifier_input_producer = { path = "core/node/tee_verifier_input_producer" }

core/bin/external_node/Cargo.toml

@@ -11,7 +11,7 @@ categories.workspace = true
 publish = false
 
 [dependencies]
-zksync_core.workspace = true
+zksync_core_leftovers.workspace = true
 zksync_commitment_generator.workspace = true
 zksync_dal.workspace = true
 zksync_db_connection.workspace = true
@@ -35,6 +35,13 @@ zksync_node_genesis.workspace = true
 zksync_node_fee_model.workspace = true
 zksync_node_db_pruner.workspace = true
 zksync_eth_sender.workspace = true
+zksync_state_keeper.workspace = true
+zksync_reorg_detector.workspace = true
+zksync_consistency_checker.workspace = true
+zksync_metadata_calculator.workspace = true
+zksync_node_sync.workspace = true
+zksync_node_api_server.workspace = true
+zksync_node_consensus.workspace = true
 vlog.workspace = true
 
 zksync_concurrency.workspace = true

core/bin/external_node/src/config/mod.rs

@@ -14,15 +14,13 @@ use zksync_config::{
     },
     ObjectStoreConfig,
 };
-use zksync_core::{
-    api_server::{
-        tx_sender::TxSenderConfig,
-        web3::{state::InternalApiConfig, Namespace},
-    },
-    temp_config_store::decode_yaml_repr,
-};
+use zksync_core_leftovers::temp_config_store::decode_yaml_repr;
 #[cfg(test)]
 use zksync_dal::{ConnectionPool, Core};
+use zksync_node_api_server::{
+    tx_sender::TxSenderConfig,
+    web3::{state::InternalApiConfig, Namespace},
+};
 use zksync_protobuf_config::proto;
 use zksync_snapshots_applier::SnapshotsApplierConfig;
 use zksync_types::{

core/bin/external_node/src/init.rs

@@ -3,9 +3,9 @@
 use std::time::Instant;
 
 use anyhow::Context as _;
-use zksync_core::sync_layer::genesis::perform_genesis_if_needed;
 use zksync_dal::{ConnectionPool, Core, CoreDal};
 use zksync_health_check::AppHealthCheck;
+use zksync_node_sync::genesis::perform_genesis_if_needed;
 use zksync_object_store::ObjectStoreFactory;
 use zksync_shared_metrics::{SnapshotRecoveryStage, APP_METRICS};
 use zksync_snapshots_applier::{SnapshotsApplierConfig, SnapshotsApplierTask};

core/bin/external_node/src/main.rs

@@ -8,40 +8,41 @@ use tokio::{
     task::{self, JoinHandle},
 };
 use zksync_block_reverter::{BlockReverter, NodeRole};
-use zksync_commitment_generator::CommitmentGenerator;
+use zksync_commitment_generator::{
+    validation_task::L1BatchCommitmentModeValidationTask, CommitmentGenerator,
+};
 use zksync_concurrency::{ctx, scope};
 use zksync_config::configs::{api::MerkleTreeApiConfig, database::MerkleTreeMode};
-use zksync_core::{
-    api_server::{
-        execution_sandbox::VmConcurrencyLimiter,
-        healthcheck::HealthCheckHandle,
-        tree::{TreeApiClient, TreeApiHttpClient},
-        tx_sender::{proxy::TxProxy, ApiContracts, TxSenderBuilder},
-        web3::{mempool_cache::MempoolCache, ApiBuilder, Namespace},
-    },
-    consensus,
-    consistency_checker::ConsistencyChecker,
-    metadata_calculator::{MetadataCalculator, MetadataCalculatorConfig},
-    reorg_detector::{self, ReorgDetector},
-    setup_sigint_handler,
-    state_keeper::{
-        seal_criteria::NoopSealer, AsyncRocksdbCache, BatchExecutor, MainBatchExecutor,
-        OutputHandler, StateKeeperPersistence, ZkSyncStateKeeper,
-    },
-    sync_layer::{
-        batch_status_updater::BatchStatusUpdater, external_io::ExternalIO, ActionQueue, SyncState,
-    },
-    utils::L1BatchCommitmentModeValidationTask,
-};
+use zksync_consistency_checker::ConsistencyChecker;
+use zksync_core_leftovers::setup_sigint_handler;
 use zksync_dal::{metrics::PostgresMetrics, ConnectionPool, Core, CoreDal};
 use zksync_db_connection::{
     connection_pool::ConnectionPoolBuilder, healthcheck::ConnectionPoolHealthCheck,
 };
 use zksync_eth_client::EthInterface;
 use zksync_health_check::{AppHealthCheck, HealthStatus, ReactiveHealthCheck};
+use zksync_metadata_calculator::{
+    api_server::{TreeApiClient, TreeApiHttpClient},
+    MetadataCalculator, MetadataCalculatorConfig,
+};
+use zksync_node_api_server::{
+    execution_sandbox::VmConcurrencyLimiter,
+    healthcheck::HealthCheckHandle,
+    tx_sender::{proxy::TxProxy, ApiContracts, TxSenderBuilder},
+    web3::{mempool_cache::MempoolCache, ApiBuilder, Namespace},
+};
+use zksync_node_consensus as consensus;
 use zksync_node_db_pruner::{DbPruner, DbPrunerConfig};
 use zksync_node_fee_model::l1_gas_price::MainNodeFeeParamsFetcher;
+use zksync_node_sync::{
+    batch_status_updater::BatchStatusUpdater, external_io::ExternalIO, ActionQueue, SyncState,
+};
+use zksync_reorg_detector::ReorgDetector;
 use zksync_state::{PostgresStorageCaches, RocksdbStorageOptions};
+use zksync_state_keeper::{
+    seal_criteria::NoopSealer, AsyncRocksdbCache, BatchExecutor, MainBatchExecutor, OutputHandler,
+    StateKeeperPersistence, ZkSyncStateKeeper,
+};
 use zksync_storage::RocksDB;
 use zksync_types::L2ChainId;
 use zksync_utils::wait_for_tasks::ManagedTasks;
@@ -964,7 +965,7 @@ async fn run_node(
         Ok(()) => {
             tracing::info!("Successfully checked no reorg compared to the main node");
         }
-        Err(reorg_detector::Error::ReorgDetected(last_correct_l1_batch)) => {
+        Err(zksync_reorg_detector::Error::ReorgDetected(last_correct_l1_batch)) => {
             tracing::info!("Reverting to l1 batch number {last_correct_l1_batch}");
             reverter.roll_back(last_correct_l1_batch).await?;
             tracing::info!("Revert successfully completed");

core/bin/genesis_generator/Cargo.toml

@@ -17,7 +17,7 @@ zksync_env_config.workspace = true
 zksync_protobuf_config.workspace = true
 zksync_utils.workspace = true
 zksync_types.workspace = true
-zksync_core.workspace = true
+zksync_core_leftovers.workspace = true
 zksync_dal.workspace = true
 zksync_contracts.workspace = true
 zksync_protobuf.workspace = true

core/bin/genesis_generator/src/main.rs

@@ -9,7 +9,7 @@ use clap::Parser;
 use serde_yaml::Serializer;
 use zksync_config::{GenesisConfig, PostgresConfig};
 use zksync_contracts::BaseSystemContracts;
-use zksync_core::temp_config_store::decode_yaml_repr;
+use zksync_core_leftovers::temp_config_store::decode_yaml_repr;
 use zksync_dal::{ConnectionPool, Core, CoreDal};
 use zksync_env_config::FromEnv;
 use zksync_node_genesis::{insert_genesis_batch, GenesisParams};

core/bin/zksync_server/Cargo.toml

@@ -18,7 +18,7 @@ zksync_protobuf_config.workspace = true
 zksync_storage.workspace = true
 zksync_utils.workspace = true
 zksync_types.workspace = true
-zksync_core.workspace = true
+zksync_core_leftovers.workspace = true
 zksync_node_genesis.workspace = true
 
 # Consensus dependenices

core/bin/zksync_server/src/config.rs

@@ -1,6 +1,6 @@
 use anyhow::Context as _;
 use zksync_config::configs::consensus::{ConsensusConfig, ConsensusSecrets};
-use zksync_core::temp_config_store::decode_yaml_repr;
+use zksync_core_leftovers::temp_config_store::decode_yaml_repr;
 use zksync_protobuf_config::proto;
 
 pub(crate) fn read_consensus_secrets() -> anyhow::Result<Option<ConsensusSecrets>> {

core/bin/zksync_server/src/main.rs

@@ -18,7 +18,7 @@ use zksync_config::{
     ApiConfig, ContractVerifierConfig, DBConfig, EthConfig, EthWatchConfig, GasAdjusterConfig,
     GenesisConfig, ObjectStoreConfig, PostgresConfig, SnapshotsCreatorConfig,
 };
-use zksync_core::{
+use zksync_core_leftovers::{
     genesis_init, initialize_components, is_genesis_needed, setup_sigint_handler,
     temp_config_store::{decode_yaml, decode_yaml_repr, Secrets, TempConfigStore},
     Component, Components,

core/lib/dal/src/blocks_dal.rs

@@ -2,6 +2,7 @@ use std::{
     collections::HashMap,
     convert::{Into, TryInto},
     ops,
+    sync::atomic::{AtomicBool, Ordering},
 };
 
 use anyhow::Context as _;
@@ -2207,6 +2208,39 @@ impl BlocksDal<'_, '_> {
         self.delete_logs_inner().await?;
         Ok(())
     }
+
+    /// Obtains a protocol version projected to be applied for the next L2 block. This is either the version used by the last
+    /// sealed L2 block, or (if there are no L2 blocks), one referenced in the snapshot recovery record.
+    pub async fn pending_protocol_version(&mut self) -> anyhow::Result<ProtocolVersionId> {
+        static WARNED_ABOUT_NO_VERSION: AtomicBool = AtomicBool::new(false);
+
+        let last_l2_block = self
+            .storage
+            .blocks_dal()
+            .get_last_sealed_l2_block_header()
+            .await?;
+        if let Some(last_l2_block) = last_l2_block {
+            return Ok(last_l2_block.protocol_version.unwrap_or_else(|| {
+                // Protocol version should be set for the most recent L2 block even in cases it's not filled
+                // for old L2 blocks, hence the warning. We don't want to rely on this assumption, so we treat
+                // the lack of it as in other similar places, replacing with the default value.
+                if !WARNED_ABOUT_NO_VERSION.fetch_or(true, Ordering::Relaxed) {
+                    tracing::warn!(
+                        "Protocol version not set for recent L2 block: {last_l2_block:?}"
+                    );
+                }
+                ProtocolVersionId::last_potentially_undefined()
+            }));
+        }
+        // No L2 blocks in the storage; use snapshot recovery information.
+        let snapshot_recovery = self
+            .storage
+            .snapshot_recovery_dal()
+            .get_applied_snapshot_status()
+            .await?
+            .context("storage contains neither L2 blocks, nor snapshot recovery info")?;
+        Ok(snapshot_recovery.protocol_version)
+    }
 }
 
 #[cfg(test)]

core/lib/dal/src/helpers.rs

@@ -0,0 +1,98 @@
+//! Various helpers for using DAL methods.
+
+use std::time::Duration;
+
+use tokio::sync::watch;
+use zksync_types::L1BatchNumber;
+
+use crate::{ConnectionPool, Core, CoreDal};
+
+/// Repeatedly polls the DB until there is an L1 batch. We may not have such a batch initially
+/// if the DB is recovered from an application-level snapshot.
+///
+/// Returns the number of the *earliest* L1 batch, or `None` if the stop signal is received.
+pub async fn wait_for_l1_batch(
+    pool: &ConnectionPool<Core>,
+    poll_interval: Duration,
+    stop_receiver: &mut watch::Receiver<bool>,
+) -> anyhow::Result<Option<L1BatchNumber>> {
+    tracing::debug!("Waiting for at least one L1 batch in db in DB");
+    loop {
+        if *stop_receiver.borrow() {
+            return Ok(None);
+        }
+
+        let mut storage = pool.connection().await?;
+        let sealed_l1_batch_number = storage.blocks_dal().get_earliest_l1_batch_number().await?;
+        drop(storage);
+
+        if let Some(number) = sealed_l1_batch_number {
+            return Ok(Some(number));
+        }
+
+        // We don't check the result: if a stop signal is received, we'll return at the start
+        // of the next iteration.
+        tokio::time::timeout(poll_interval, stop_receiver.changed())
+            .await
+            .ok();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use zksync_contracts::BaseSystemContractsHashes;
+    use zksync_types::{block::L1BatchHeader, ProtocolVersion, ProtocolVersionId, H256};
+
+    use super::*;
+    use crate::{ConnectionPool, Core, CoreDal};
+
+    #[tokio::test]
+    async fn waiting_for_l1_batch_success() {
+        let pool = ConnectionPool::<Core>::test_pool().await;
+        let (_stop_sender, mut stop_receiver) = watch::channel(false);
+
+        let pool_copy = pool.clone();
+        tokio::spawn(async move {
+            tokio::time::sleep(Duration::from_millis(25)).await;
+            let mut conn = pool_copy.connection().await.unwrap();
+            conn.protocol_versions_dal()
+                .save_protocol_version_with_tx(&ProtocolVersion::default())
+                .await
+                .unwrap();
+            let header = L1BatchHeader::new(
+                L1BatchNumber(0),
+                100,
+                BaseSystemContractsHashes {
+                    bootloader: H256::repeat_byte(1),
+                    default_aa: H256::repeat_byte(42),
+                },
+                ProtocolVersionId::latest(),
+            );
+            conn.blocks_dal()
+                .insert_mock_l1_batch(&header)
+                .await
+                .unwrap();
+        });
+
+        let l1_batch = wait_for_l1_batch(&pool, Duration::from_millis(10), &mut stop_receiver)
+            .await
+            .unwrap();
+        assert_eq!(l1_batch, Some(L1BatchNumber(0)));
+    }
+
+    #[tokio::test]
+    async fn waiting_for_l1_batch_cancellation() {
+        let pool = ConnectionPool::<Core>::test_pool().await;
+        let (stop_sender, mut stop_receiver) = watch::channel(false);
+
+        tokio::spawn(async move {
+            tokio::time::sleep(Duration::from_millis(25)).await;
+            stop_sender.send_replace(true);
+        });
+
+        let l1_batch = wait_for_l1_batch(&pool, Duration::from_secs(30), &mut stop_receiver)
+            .await
+            .unwrap();
+        assert_eq!(l1_batch, None);
+    }
+}

core/lib/dal/src/lib.rs

@@ -35,6 +35,7 @@ pub mod eth_sender_dal;
 pub mod events_dal;
 pub mod events_web3_dal;
 pub mod factory_deps_dal;
+pub mod helpers;
 pub mod metrics;
 mod models;
 pub mod proof_generation_dal;

core/lib/zksync_core/Cargo.toml → core/lib/zksync_core_leftovers/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "zksync_core"
+name = "zksync_core_leftovers"
 version = "0.1.0"
 edition.workspace = true
 authors.workspace = true
@@ -9,7 +9,7 @@ license.workspace = true
 keywords.workspace = true
 categories.workspace = true
 
-links = "zksync_core_proto"
+links = "zksync_core_leftovers_proto"
 
 [dependencies]
 vise.workspace = true
@@ -51,6 +51,13 @@ zksync_house_keeper.workspace = true
 zksync_node_genesis.workspace = true
 zksync_eth_sender.workspace = true
 zksync_node_fee_model.workspace = true
+zksync_state_keeper.workspace = true
+zksync_metadata_calculator.workspace = true
+zksync_node_sync.workspace = true
+zksync_node_consensus.workspace = true
+zksync_contract_verification_server.workspace = true
+zksync_node_api_server.workspace = true
+zksync_tee_verifier_input_producer.workspace = true
 multivm.workspace = true
 
 # Consensus dependenices