feat: Implemented 1 validator consensus for the main node (#554)

## What ❔

Implemented 1 validator consensus for the main node.
It is NOT enabled yet, corresponding configuration still needs to be
provided.
It is responsible for populating consensus column in the Miniblocks
table.
Consensus will be running asynchronously for now, not affecting the
miniblock production at all.

Fixes BFT-388.

## Why ❔

Reaching a consensus with just 1 validator doesn't provide much value,
however we want to start running the consensus code in production (in a
"shadow" mode) to make sure it works as expected before we start relying
on it.

core/lib/dal/Cargo.toml

@@ -18,8 +18,9 @@ zksync_system_constants = { path = "../constants" }
 zksync_contracts = { path = "../contracts" }
 zksync_types = { path = "../types" }
 zksync_health_check = { path = "../health_check" }
-zksync_consensus_roles = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "da015d4c94b19962bc11622b6cc256e214256555" }
-zksync_protobuf = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "da015d4c94b19962bc11622b6cc256e214256555" }
+zksync_consensus_roles = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "49b1a98f80d0e9f74fdceadece4283e745c71599" }
+zksync_consensus_storage = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "49b1a98f80d0e9f74fdceadece4283e745c71599" }
+zksync_protobuf = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "49b1a98f80d0e9f74fdceadece4283e745c71599" }
 
 itertools = "0.10.1"
 thiserror = "1.0"
@@ -53,4 +54,4 @@ tracing = "0.1"
 assert_matches = "1.5.0"
 
 [build-dependencies]
-zksync_protobuf_build = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "da015d4c94b19962bc11622b6cc256e214256555" }
+zksync_protobuf_build = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "49b1a98f80d0e9f74fdceadece4283e745c71599" }

core/lib/dal/migrations/20231128123456_create_consensus_replica_state_table.down.sql

@@ -0,0 +1 @@
+DROP TABLE consensus_replica_state;

core/lib/dal/migrations/20231128123456_create_consensus_replica_state_table.up.sql

@@ -0,0 +1,6 @@
+CREATE TABLE IF NOT EXISTS consensus_replica_state (
+    state JSONB NOT NULL,
+    -- artificial primary key ensuring that the table contains at most 1 row.
+    fake_key BOOLEAN PRIMARY KEY,
+    CHECK (fake_key)
+);

core/lib/dal/sqlx-data.json

@@ -13,6 +13,24 @@
     },
     "query": "UPDATE proof_generation_details SET status=$1, updated_at = now() WHERE l1_batch_number = $2"
   },
+  "002a901f2802beb48e4d15437fba9a04885a5e4f8f17089a09f31edfd7b3d0bb": {
+    "describe": {
+      "columns": [
+        {
+          "name": "number",
+          "ordinal": 0,
+          "type_info": "Int8"
+        }
+      ],
+      "nullable": [
+        false
+      ],
+      "parameters": {
+        "Left": []
+      }
+    },
+    "query": "SELECT number FROM miniblocks WHERE consensus IS NOT NULL ORDER BY number DESC LIMIT 1"
+  },
   "00bd80fd83aff559d8d9232c2e98a12a1dd2c8f31792cd915e2cf11f28e583b7": {
     "describe": {
       "columns": [
@@ -710,6 +728,24 @@
     },
     "query": "SELECT l1_batch_number, factory_deps_filepath, storage_logs_filepaths FROM snapshots WHERE l1_batch_number = $1"
   },
+  "166dcd8d504ba3f52a9e44a05305ed00954ab9b5302be4bb5ab05dfd2272afca": {
+    "describe": {
+      "columns": [
+        {
+          "name": "state!",
+          "ordinal": 0,
+          "type_info": "Jsonb"
+        }
+      ],
+      "nullable": [
+        false
+      ],
+      "parameters": {
+        "Left": []
+      }
+    },
+    "query": "SELECT state as \"state!\" FROM consensus_replica_state WHERE fake_key"
+  },
   "16bca6f4258ff3db90a26a8550c5fc35e666fb698960486528fceba3e452fd62": {
     "describe": {
       "columns": [
@@ -10486,6 +10522,18 @@
     },
     "query": "UPDATE basic_witness_input_producer_jobs SET status = $1, attempts = attempts + 1, updated_at = now(), processing_started_at = now() WHERE l1_batch_number = ( SELECT l1_batch_number FROM basic_witness_input_producer_jobs WHERE status = $2 OR (status = $1 AND processing_started_at < now() - $4::interval) OR (status = $3 AND attempts < $5) ORDER BY l1_batch_number ASC LIMIT 1 FOR UPDATE SKIP LOCKED ) RETURNING basic_witness_input_producer_jobs.l1_batch_number"
   },
+  "e577743852337c926e1566c46f4cbab5e6ab1409921fa8e25c9dfa7dfa03daae": {
+    "describe": {
+      "columns": [],
+      "nullable": [],
+      "parameters": {
+        "Left": [
+          "Jsonb"
+        ]
+      }
+    },
+    "query": "INSERT INTO consensus_replica_state(fake_key,state) VALUES(true,$1) ON CONFLICT (fake_key) DO UPDATE SET state = excluded.state"
+  },
   "e626aa2efb6ba875a12f2b4e37b0ba8052810e73fa5e2d3280f747f7b89b956f": {
     "describe": {
       "columns": [],
@@ -10513,6 +10561,26 @@
     },
     "query": "UPDATE eth_txs SET gas_used = $1, confirmed_eth_tx_history_id = $2 WHERE id = $3"
   },
+  "e8629da5d4269a565f1043969d29df2ccfbebafdce96d58c601860f30d61b4a0": {
+    "describe": {
+      "columns": [
+        {
+          "name": "count!",
+          "ordinal": 0,
+          "type_info": "Int8"
+        }
+      ],
+      "nullable": [
+        null
+      ],
+      "parameters": {
+        "Left": [
+          "Int8"
+        ]
+      }
+    },
+    "query": "SELECT COUNT(*) as \"count!\"  FROM miniblocks WHERE number = $1 AND consensus IS NOT NULL"
+  },
   "e8988deed66ad9d10be89e89966082aeb920c5dc91eb5fad16bd0d3118708c2e": {
     "describe": {
       "columns": [

core/lib/dal/src/blocks_dal.rs

@@ -396,6 +396,30 @@ impl BlocksDal<'_, '_> {
         .await?;
         Ok(())
     }
+
+    /// Fetches the number of the last miniblock with consensus fields set.
+    /// Miniblocks with Consensus fields set constitute a prefix of sealed miniblocks,
+    /// so it is enough to traverse the miniblocks in descending order to find the last
+    /// with consensus fields.
+    ///
+    /// If better efficiency is needed we can add an index on "miniblocks without consensus fields".
+    pub async fn get_last_miniblock_number_with_consensus_fields(
+        &mut self,
+    ) -> anyhow::Result<Option<MiniblockNumber>> {
+        let Some(row) = sqlx::query!("SELECT number FROM miniblocks WHERE consensus IS NOT NULL ORDER BY number DESC LIMIT 1")
+            .fetch_optional(self.storage.conn())
+            .await? else { return Ok(None) };
+        Ok(Some(MiniblockNumber(row.number.try_into()?)))
+    }
+
+    /// Checks whether the specified miniblock has consensus field set.
+    pub async fn has_consensus_fields(&mut self, number: MiniblockNumber) -> sqlx::Result<bool> {
+        Ok(sqlx::query!("SELECT COUNT(*) as \"count!\"  FROM miniblocks WHERE number = $1 AND consensus IS NOT NULL", number.0 as i64)
+            .fetch_one(self.storage.conn())
+            .await?
+            .count > 0)
+    }
+
     /// Sets consensus-related fields for the specified miniblock.
     pub async fn set_miniblock_consensus_fields(
         &mut self,

core/lib/dal/src/consensus_dal.rs

@@ -0,0 +1,59 @@
+use zksync_consensus_storage::ReplicaState;
+
+use crate::StorageProcessor;
+
+#[derive(Debug)]
+pub struct ConsensusDal<'a, 'c> {
+    pub storage: &'a mut StorageProcessor<'c>,
+}
+
+impl ConsensusDal<'_, '_> {
+    pub async fn replica_state(&mut self) -> anyhow::Result<Option<ReplicaState>> {
+        let Some(row) =
+            sqlx::query!("SELECT state as \"state!\" FROM consensus_replica_state WHERE fake_key")
+                .fetch_optional(self.storage.conn())
+                .await?
+        else {
+            return Ok(None);
+        };
+        Ok(Some(zksync_protobuf::serde::deserialize(row.state)?))
+    }
+
+    pub async fn put_replica_state(&mut self, state: &ReplicaState) -> sqlx::Result<()> {
+        let state =
+            zksync_protobuf::serde::serialize(state, serde_json::value::Serializer).unwrap();
+        sqlx::query!("INSERT INTO consensus_replica_state(fake_key,state) VALUES(true,$1) ON CONFLICT (fake_key) DO UPDATE SET state = excluded.state", state)
+            .execute(self.storage.conn())
+            .await?;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use rand::Rng as _;
+    use zksync_consensus_storage::ReplicaState;
+
+    use crate::ConnectionPool;
+
+    #[tokio::test]
+    async fn replica_state_read_write() {
+        let pool = ConnectionPool::test_pool().await;
+        let mut conn = pool.access_storage().await.unwrap();
+        assert!(conn
+            .consensus_dal()
+            .replica_state()
+            .await
+            .unwrap()
+            .is_none());
+        let rng = &mut rand::thread_rng();
+        for _ in 0..10 {
+            let want: ReplicaState = rng.gen();
+            conn.consensus_dal().put_replica_state(&want).await.unwrap();
+            assert_eq!(
+                Some(want),
+                conn.consensus_dal().replica_state().await.unwrap()
+            );
+        }
+    }
+}

core/lib/dal/src/lib.rs

@@ -7,8 +7,8 @@ pub use crate::connection::ConnectionPool;
 use crate::{
     accounts_dal::AccountsDal, basic_witness_input_producer_dal::BasicWitnessInputProducerDal,
     blocks_dal::BlocksDal, blocks_web3_dal::BlocksWeb3Dal, connection::holder::ConnectionHolder,
-    contract_verification_dal::ContractVerificationDal, eth_sender_dal::EthSenderDal,
-    events_dal::EventsDal, events_web3_dal::EventsWeb3Dal,
+    consensus_dal::ConsensusDal, contract_verification_dal::ContractVerificationDal,
+    eth_sender_dal::EthSenderDal, events_dal::EventsDal, events_web3_dal::EventsWeb3Dal,
     fri_gpu_prover_queue_dal::FriGpuProverQueueDal,
     fri_proof_compressor_dal::FriProofCompressorDal,
     fri_protocol_versions_dal::FriProtocolVersionsDal, fri_prover_dal::FriProverDal,
@@ -31,6 +31,7 @@ pub mod basic_witness_input_producer_dal;
 pub mod blocks_dal;
 pub mod blocks_web3_dal;
 pub mod connection;
+pub mod consensus_dal;
 pub mod contract_verification_dal;
 pub mod eth_sender_dal;
 pub mod events_dal;
@@ -145,6 +146,10 @@ impl<'a> StorageProcessor<'a> {
         BlocksWeb3Dal { storage: self }
     }
 
+    pub fn consensus_dal(&mut self) -> ConsensusDal<'_, 'a> {
+        ConsensusDal { storage: self }
+    }
+
     pub fn eth_sender_dal(&mut self) -> EthSenderDal<'_, 'a> {
         EthSenderDal { storage: self }
     }

core/lib/types/Cargo.toml

@@ -21,8 +21,8 @@ codegen = { git = "https://github.com/matter-labs/solidity_plonk_verifier.git",
 zkevm_test_harness = { git = "https://github.com/matter-labs/era-zkevm_test_harness.git", branch = "v1.3.3" }
 zk_evm_1_4_0 = { git = "https://github.com/matter-labs/era-zk_evm.git", branch = "v1.4.0", package = "zk_evm" }
 zk_evm = { git = "https://github.com/matter-labs/era-zk_evm.git", tag = "v1.3.3-rc2" }
-zksync_consensus_roles = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "da015d4c94b19962bc11622b6cc256e214256555" }
-zksync_protobuf = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "da015d4c94b19962bc11622b6cc256e214256555" }
+zksync_consensus_roles = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "49b1a98f80d0e9f74fdceadece4283e745c71599" }
+zksync_protobuf = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "49b1a98f80d0e9f74fdceadece4283e745c71599" }
 
 anyhow = "1.0.75"
 chrono = { version = "0.4", features = ["serde"] }
@@ -52,4 +52,4 @@ tokio = { version = "1", features = ["rt", "macros"] }
 serde_with = { version = "1", features = ["hex"] }
 
 [build-dependencies]
-zksync_protobuf_build = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "da015d4c94b19962bc11622b6cc256e214256555" }
+zksync_protobuf_build = { version = "0.1.0", git = "https://github.com/matter-labs/era-consensus.git", rev = "49b1a98f80d0e9f74fdceadece4283e745c71599" }