[WIP] feat(consensus): committees data extractor (BFT-443)

Cargo.toml

@@ -211,16 +211,16 @@ zk_evm_1_4_1 = { package = "zk_evm", version = "0.141.0" }
 zk_evm_1_5_0 = { package = "zk_evm", version = "0.150.0" }
 
 # Consensus dependencies.
-zksync_concurrency = "=0.1.0-rc.4"
-zksync_consensus_bft = "=0.1.0-rc.4"
-zksync_consensus_crypto = "=0.1.0-rc.4"
-zksync_consensus_executor = "=0.1.0-rc.4"
-zksync_consensus_network = "=0.1.0-rc.4"
-zksync_consensus_roles = "=0.1.0-rc.4"
-zksync_consensus_storage = "=0.1.0-rc.4"
-zksync_consensus_utils = "=0.1.0-rc.4"
-zksync_protobuf = "=0.1.0-rc.4"
-zksync_protobuf_build = "=0.1.0-rc.4"
+zksync_concurrency = { git = "https://github.com/matter-labs/era-consensus.git", branch = "update_committee_types" }
+zksync_consensus_bft = { git = "https://github.com/matter-labs/era-consensus.git", branch = "update_committee_types" }
+zksync_consensus_crypto = { git = "https://github.com/matter-labs/era-consensus.git", branch = "update_committee_types" }
+zksync_consensus_executor = { git = "https://github.com/matter-labs/era-consensus.git", branch = "update_committee_types" }
+zksync_consensus_network = { git = "https://github.com/matter-labs/era-consensus.git", branch = "update_committee_types" }
+zksync_consensus_roles = { git = "https://github.com/matter-labs/era-consensus.git", branch = "update_committee_types" }
+zksync_consensus_storage = { git = "https://github.com/matter-labs/era-consensus.git", branch = "update_committee_types" }
+zksync_consensus_utils = { git = "https://github.com/matter-labs/era-consensus.git", branch = "update_committee_types" }
+zksync_protobuf = { git = "https://github.com/matter-labs/era-consensus.git", branch = "update_committee_types" }
+zksync_protobuf_build = { git = "https://github.com/matter-labs/era-consensus.git", branch = "update_committee_types" }
 
 # "Local" dependencies
 zksync_multivm = { version = "0.1.0", path = "core/lib/multivm" }

core/lib/dal/migrations/20240725194916_l1_batches_consensus_committees.down.sql

@@ -0,0 +1,6 @@
+ALTER TABLE l1_batches_consensus
+DROP COLUMN validator_committee,
+DROP COLUMN attester_committee;
+
+ALTER TABLE l1_batches_consensus
+ALTER COLUMN certificate SET NOT NULL; 

core/lib/dal/migrations/20240725194916_l1_batches_consensus_committees.up.sql

@@ -0,0 +1,12 @@
+ALTER TABLE l1_batches_consensus
+ADD validator_committee JSONB NULL,
+ADD attester_committee JSONB NULL;
+
+ALTER TABLE l1_batches_consensus
+ALTER COLUMN certificate DROP NOT NULL;
+
+--ALTER TABLE l1_batches_consensus DROP CONSTRAINT l1_batches_consensus_certificate_check;
+--
+--ALTER TABLE l1_batches_consensus ADD CONSTRAINT l1_batches_consensus_certificate_check
+--CHECK (certificate IS NULL OR ((certificate->'message'->>'number')::numeric = l1_batch_number));
+--

core/lib/dal/src/consensus_dal.rs

@@ -1,12 +1,18 @@
 use anyhow::Context as _;
 use bigdecimal::Zero as _;
-use zksync_consensus_roles::{attester, validator};
+use zksync_consensus_roles::{
+    attester,
+    attester::{AttesterCommittee, BatchNumber},
+    validator,
+    validator::{ValidatorCommittee, WeightedValidator},
+};
 use zksync_consensus_storage::{BlockStoreState, ReplicaState};
 use zksync_db_connection::{
     connection::Connection,
     error::{DalError, DalResult, SqlxContext},
     instrument::{InstrumentExt, Instrumented},
 };
+use zksync_protobuf::ProtoFmt;
 use zksync_types::L2BlockNumber;
 
 pub use crate::consensus::Payload;
@@ -317,7 +323,39 @@ impl ConsensusDal<'_, '_> {
         else {
             return Ok(None);
         };
-        Ok(Some(zksync_protobuf::serde::deserialize(row.certificate)?))
+        Ok(Some(zksync_protobuf::serde::deserialize(
+            row.certificate.unwrap(),
+        )?))
+    }
+
+    pub async fn batch_committees(
+        &mut self,
+        batch_number: attester::BatchNumber,
+    ) -> anyhow::Result<Option<(validator::ValidatorCommittee, attester::AttesterCommittee)>> {
+        let Some(row) = sqlx::query!(
+            r#"
+            SELECT
+                validator_committee,
+                attester_committee
+            FROM
+                l1_batches_consensus
+            WHERE
+                l1_batch_number = $1
+            "#,
+            i64::try_from(batch_number.0)?
+        )
+        .instrument("batch_committees")
+        .report_latency()
+        .fetch_optional(self.storage)
+        .await?
+        else {
+            return Ok(None);
+        };
+
+        Ok(Some((
+            zksync_protobuf::serde::deserialize(row.validator_committee.unwrap())?,
+            zksync_protobuf::serde::deserialize(row.attester_committee.unwrap())?,
+        )))
     }
 
     /// Fetches a range of L2 blocks from storage and converts them to `Payload`s.
@@ -402,6 +440,48 @@ impl ConsensusDal<'_, '_> {
         Ok(())
     }
 
+    pub async fn insert_batch_committees(
+        &mut self,
+        number: BatchNumber,
+        validator_committee: ValidatorCommittee,
+        attester_committee: AttesterCommittee,
+    ) -> anyhow::Result<()> {
+        let res = sqlx::query!(
+            r#"
+            INSERT INTO
+                l1_batches_consensus (
+                    l1_batch_number,
+                    certificate,
+                    validator_committee,
+                    attester_committee,
+                    created_at,
+                    updated_at
+                )
+            VALUES
+                ($1, NULL, $2, $3, NOW(), NOW())
+            ON CONFLICT (l1_batch_number) DO
+            UPDATE
+            SET
+                validator_committee = EXCLUDED.validator_committee,
+                attester_committee = EXCLUDED.attester_committee,
+                updated_at = NOW();
+            "#,
+            i64::try_from(number.0).context("overflow")?.into(),
+            zksync_protobuf::serde::serialize(&validator_committee, serde_json::value::Serializer)
+                .unwrap(),
+            zksync_protobuf::serde::serialize(&attester_committee, serde_json::value::Serializer)
+                .unwrap(),
+        )
+        .instrument("insert_batch_committees")
+        .report_latency()
+        .execute(self.storage)
+        .await?;
+        if res.rows_affected().is_zero() {
+            tracing::debug!(l1_batch_number = ?number, "duplicate batch certificate");
+        }
+        Ok(())
+    }
+
     /// Inserts a certificate for the L1 batch.
     /// Noop if a certificate for the same L1 batch is already present.
     /// No verification is performed - it cannot be performed due to circular dependency on
@@ -413,10 +493,21 @@ impl ConsensusDal<'_, '_> {
         let res = sqlx::query!(
             r#"
             INSERT INTO
-                l1_batches_consensus (l1_batch_number, certificate, created_at, updated_at)
+                l1_batches_consensus (
+                    l1_batch_number,
+                    certificate,
+                    validator_committee,
+                    attester_committee,
+                    created_at,
+                    updated_at
+                )
             VALUES
-                ($1, $2, NOW(), NOW())
-            ON CONFLICT (l1_batch_number) DO NOTHING
+                ($1, $2, NULL, NULL, NOW(), NOW())
+            ON CONFLICT (l1_batch_number) DO
+            UPDATE
+            SET
+                certificate = EXCLUDED.certificate,
+                updated_at = NOW();
             "#,
             i64::try_from(cert.message.number.0).context("overflow")?,
             // Unwrap is ok, because serialization should always succeed.

core/node/consensus/src/config.rs

@@ -10,7 +10,7 @@ use zksync_config::{
 };
 use zksync_consensus_crypto::{Text, TextFmt};
 use zksync_consensus_executor as executor;
-use zksync_consensus_roles::{attester, node, validator};
+use zksync_consensus_roles::{attester, node, validator, validator::Signature};
 
 fn read_secret_text<T: TextFmt>(text: Option<&Secret<String>>) -> anyhow::Result<Option<T>> {
     text.map(|text| Text::new(text.expose_secret()).decode())
@@ -63,6 +63,7 @@ impl GenesisSpec {
                 Ok(validator::WeightedValidator {
                     key: Text::new(&v.key.0).decode().context("key").context(i)?,
                     weight: v.weight,
+                    pop: Signature::decode(Text::new("")).unwrap(), // TODO(moshababo): implement
                 })
             })
             .collect::<anyhow::Result<_>>()

core/node/consensus/src/storage/connection.rs

@@ -1,7 +1,12 @@
 use anyhow::Context as _;
 use zksync_concurrency::{ctx, error::Wrap as _, time};
 use zksync_consensus_crypto::keccak256::Keccak256;
-use zksync_consensus_roles::{attester, validator};
+use zksync_consensus_roles::{
+    attester,
+    attester::{AttesterCommittee, BatchNumber},
+    validator,
+    validator::ValidatorCommittee,
+};
 use zksync_consensus_storage::{self as storage, BatchStoreState};
 use zksync_dal::{consensus_dal, consensus_dal::Payload, Core, CoreDal, DalError};
 use zksync_l1_contract_interface::i_executor::structures::StoredBatchInfo;
@@ -138,6 +143,23 @@ impl<'a> Connection<'a> {
             .map_err(E::Other)?)
     }
 
+    /// Wrapper for `consensus_dal().insert_batch_committees()`.
+    pub async fn insert_batch_committees(
+        &mut self,
+        ctx: &ctx::Ctx,
+        number: BatchNumber,
+        validator_committee: ValidatorCommittee,
+        attester_committee: AttesterCommittee,
+    ) -> ctx::Result<()> {
+        ctx.wait(self.0.consensus_dal().insert_batch_committees(
+            number,
+            validator_committee,
+            attester_committee,
+        ))
+        .await??;
+        Ok(())
+    }
+
     /// Wrapper for `consensus_dal().replica_state()`.
     pub async fn replica_state(&mut self, ctx: &ctx::Ctx) -> ctx::Result<storage::ReplicaState> {
         Ok(ctx
@@ -339,6 +361,18 @@ impl<'a> Connection<'a> {
             .context("batch_certificate()")?)
     }
 
+    /// Wrapper for `consensus_dal().batch_committees()`.
+    pub async fn batch_committees(
+        &mut self,
+        ctx: &ctx::Ctx,
+        batch_number: attester::BatchNumber,
+    ) -> ctx::Result<Option<(validator::ValidatorCommittee, attester::AttesterCommittee)>> {
+        Ok(ctx
+            .wait(self.0.consensus_dal().batch_committees(batch_number))
+            .await?
+            .context("batch_committees()")?)
+    }
+
     /// Wrapper for `blocks_dal().get_l2_block_range_of_l1_batch()`.
     pub async fn get_l2_block_range_of_l1_batch(
         &mut self,