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

core/lib/dal/Cargo.toml

@@ -20,6 +20,7 @@ zksync_contracts.workspace = true
 zksync_types.workspace = true
 zksync_consensus_roles.workspace = true
 zksync_consensus_storage.workspace = true
+zksync_consensus_crypto.workspace = true 
 zksync_protobuf.workspace = true
 zksync_db_connection.workspace = true
 

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

@@ -0,0 +1,5 @@
+ALTER TABLE l1_batches_consensus
+DROP COLUMN 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,11 @@
+ALTER TABLE l1_batches_consensus
+ADD committee JSONB NOT 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/mod.rs

@@ -4,7 +4,8 @@ pub mod proto;
 mod tests;
 
 use anyhow::{anyhow, Context as _};
-use zksync_consensus_roles::validator;
+use zksync_consensus_crypto::ByteFmt;
+use zksync_consensus_roles::{attester, validator};
 use zksync_protobuf::{required, ProtoFmt, ProtoRepr};
 use zksync_types::{
     abi, ethabi,
@@ -439,3 +440,117 @@ impl ProtoRepr for proto::Transaction {
         }
     }
 }
+
+#[derive(Debug, PartialEq)]
+pub struct Validator {
+    /// Validator public key.
+    pub pub_key: validator::PublicKey,
+    /// Validator weight inside a Committee.
+    pub weight: validator::Weight,
+    /// Proof-of-possession (PoP) of the validator's public key (a signature over the public key)
+    pub proof_of_possession: validator::Signature,
+}
+
+impl ProtoFmt for Validator {
+    type Proto = proto::Validator;
+
+    fn read(r: &Self::Proto) -> anyhow::Result<Self> {
+        Ok(Self {
+            pub_key: required(&r.pub_key)
+                .map(|x| validator::PublicKey::decode(x))
+                .context("pub_key")??,
+            weight: *required(&r.weight).context("weight")?,
+            proof_of_possession: required(&r.proof_of_possession)
+                .map(|x| validator::Signature::decode(x))
+                .context("proof_of_possession")??,
+        })
+    }
+
+    fn build(&self) -> Self::Proto {
+        Self::Proto {
+            pub_key: Some(self.pub_key.encode()),
+            weight: Some(self.weight),
+            proof_of_possession: Some(self.proof_of_possession.encode()),
+        }
+    }
+}
+
+#[derive(Debug, PartialEq)]
+pub struct ValidatorCommittee {
+    pub members: Vec<Validator>,
+}
+
+impl ProtoFmt for ValidatorCommittee {
+    type Proto = proto::ValidatorCommittee;
+
+    fn read(r: &Self::Proto) -> anyhow::Result<Self> {
+        Ok(Self {
+            members: r
+                .members
+                .iter()
+                .map(ProtoFmt::read)
+                .collect::<Result<_, _>>()
+                .context("members")?,
+        })
+    }
+
+    fn build(&self) -> Self::Proto {
+        Self::Proto {
+            members: self.members.iter().map(|x| x.build()).collect(),
+        }
+    }
+}
+
+#[derive(Debug, PartialEq, Clone)]
+pub struct Attester {
+    /// Attester public key.
+    pub pub_key: attester::PublicKey,
+    /// Attester weight inside a Committee.
+    pub weight: attester::Weight,
+}
+
+impl ProtoFmt for Attester {
+    type Proto = proto::Attester;
+
+    fn read(r: &Self::Proto) -> anyhow::Result<Self> {
+        Ok(Self {
+            pub_key: required(&r.pub_key)
+                .map(|x| attester::PublicKey::decode(x))
+                .context("pub_key")??,
+            weight: *required(&r.weight).context("weight")?,
+        })
+    }
+
+    fn build(&self) -> Self::Proto {
+        Self::Proto {
+            pub_key: Some(self.pub_key.encode()),
+            weight: Some(self.weight),
+        }
+    }
+}
+
+#[derive(Debug, PartialEq, Default, Clone)]
+pub struct AttesterCommittee {
+    pub members: Vec<Attester>,
+}
+
+impl ProtoFmt for AttesterCommittee {
+    type Proto = proto::AttesterCommittee;
+
+    fn read(r: &Self::Proto) -> anyhow::Result<Self> {
+        Ok(Self {
+            members: r
+                .members
+                .iter()
+                .map(ProtoFmt::read)
+                .collect::<Result<_, _>>()
+                .context("members")?,
+        })
+    }
+
+    fn build(&self) -> Self::Proto {
+        Self::Proto {
+            members: self.members.iter().map(|x| x.build()).collect(),
+        }
+    }
+}

core/lib/dal/src/consensus/proto/mod.proto

@@ -114,3 +114,22 @@ message PaymasterParams {
   optional bytes paymaster_address = 1; // required; H160
   optional bytes paymaster_input = 2; // required
 }
+
+message Validator {
+  optional bytes pub_key = 1; // required
+  optional uint64 weight = 2; // required
+  optional bytes proof_of_possession = 3; // required
+}
+
+message Attester {
+  optional bytes pub_key = 1; // required
+  optional uint64 weight = 2; // required
+}
+
+message ValidatorCommittee {
+   repeated Validator members = 1; // required
+}
+
+message AttesterCommittee {
+   repeated Attester members = 1; // required
+}

core/lib/dal/src/consensus_dal.rs

@@ -1,16 +1,19 @@
 use anyhow::Context as _;
 use bigdecimal::Zero as _;
-use zksync_consensus_roles::{attester, validator};
+use zksync_consensus_roles::{
+    attester, attester::BatchNumber, validator, validator::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;
-use crate::{Core, CoreDal};
+pub use crate::consensus::{Attester, AttesterCommittee, Payload, Validator, ValidatorCommittee};
+use crate::{consensus, Core, CoreDal};
 
 /// Storage access methods for `zksync_core::consensus` module.
 #[derive(Debug)]
@@ -317,7 +320,36 @@ impl ConsensusDal<'_, '_> {
         else {
             return Ok(None);
         };
-        Ok(Some(zksync_protobuf::serde::deserialize(row.certificate)?))
+        let Some(certificate) = row.certificate else {
+            return Ok(None);
+        };
+        Ok(Some(zksync_protobuf::serde::deserialize(certificate)?))
+    }
+
+    pub async fn batch_committee(
+        &mut self,
+        batch_number: attester::BatchNumber,
+    ) -> anyhow::Result<Option<AttesterCommittee>> {
+        let Some(row) = sqlx::query!(
+            r#"
+            SELECT
+                committee
+            FROM
+                l1_batches_consensus
+            WHERE
+                l1_batch_number = $1
+            "#,
+            i64::try_from(batch_number.0)?
+        )
+        .instrument("batch_committee")
+        .report_latency()
+        .fetch_optional(self.storage)
+        .await?
+        else {
+            return Ok(None);
+        };
+
+        Ok(Some(zksync_protobuf::serde::deserialize(row.committee)?))
     }
 
     /// Fetches a range of L2 blocks from storage and converts them to `Payload`s.
@@ -402,6 +434,32 @@ impl ConsensusDal<'_, '_> {
         Ok(())
     }
 
+    pub async fn insert_batch_committee(
+        &mut self,
+        number: BatchNumber,
+        committee: AttesterCommittee,
+    ) -> anyhow::Result<()> {
+        let res = sqlx::query!(
+            r#"
+            INSERT INTO
+                l1_batches_consensus (l1_batch_number, certificate, committee, created_at, updated_at)
+            VALUES
+                ($1, NULL, $2, NOW(), NOW())
+            ON CONFLICT (l1_batch_number) DO NOTHING
+            "#,
+            i64::try_from(number.0).context("overflow")?.into(),
+            zksync_protobuf::serde::serialize(&committee, serde_json::value::Serializer).unwrap(),
+        )
+        .instrument("insert_batch_committee")
+        .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
@@ -412,11 +470,12 @@ impl ConsensusDal<'_, '_> {
     ) -> anyhow::Result<()> {
         let res = sqlx::query!(
             r#"
-            INSERT INTO
-                l1_batches_consensus (l1_batch_number, certificate, created_at, updated_at)
-            VALUES
-                ($1, $2, NOW(), NOW())
-            ON CONFLICT (l1_batch_number) DO NOTHING
+            UPDATE l1_batches_consensus
+            SET
+                certificate = $2,
+                updated_at = NOW()
+            WHERE
+                l1_batch_number = $1
             "#,
             i64::try_from(cert.message.number.0).context("overflow")?,
             // Unwrap is ok, because serialization should always succeed.
@@ -443,6 +502,8 @@ impl ConsensusDal<'_, '_> {
                 MAX(l1_batch_number) AS "number"
             FROM
                 l1_batches_consensus
+            WHERE
+                certificate IS NOT NULL
             "#
         )
         .instrument("get_last_batch_certificate_number")
@@ -491,6 +552,45 @@ impl ConsensusDal<'_, '_> {
         // and doesn't have prunning enabled.
         Ok(attester::BatchNumber(0))
     }
+
+    pub async fn get_last_batch_committee_number(
+        &mut self,
+    ) -> anyhow::Result<Option<attester::BatchNumber>> {
+        let row = sqlx::query!(
+            r#"
+            SELECT
+                MAX(l1_batch_number) AS "number"
+            FROM
+                l1_batches_consensus
+            "#
+        )
+        .instrument("get_last_batch_committee_number")
+        .report_latency()
+        .fetch_one(self.storage)
+        .await?;
+
+        let Some(n) = row.number else {
+            return Ok(None);
+        };
+        Ok(Some(attester::BatchNumber(
+            n.try_into().context("overflow")?,
+        )))
+    }
+
+    pub async fn next_batch_to_extract_committee(
+        &mut self,
+    ) -> anyhow::Result<attester::BatchNumber> {
+        // First batch that we don't have a committee for.
+        if let Some(last) = self
+            .get_last_batch_committee_number()
+            .await
+            .context("get_last_batch_certificate_number()")?
+        {
+            return Ok(last + 1);
+        }
+
+        Ok(attester::BatchNumber(0))
+    }
 }
 
 #[cfg(test)]

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())