refactor: made attestation start from consensus genesis instead of th…

…e last sealed batch (#2539)

Stable starting point prevents race conditions.

core/lib/dal/build.rs

@@ -3,7 +3,7 @@ fn main() {
     zksync_protobuf_build::Config {
         input_root: "src/consensus/proto".into(),
         proto_root: "zksync/dal".into(),
-        dependencies: vec![],
+        dependencies: vec!["::zksync_consensus_roles::proto".parse().unwrap()],
         protobuf_crate: "::zksync_protobuf".parse().unwrap(),
         is_public: true,
     }

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

@@ -1,11 +1,13 @@
 pub mod proto;
 
+#[cfg(test)]
+mod testonly;
 #[cfg(test)]
 mod tests;
 
 use anyhow::{anyhow, Context as _};
-use zksync_consensus_roles::validator;
-use zksync_protobuf::{required, ProtoFmt, ProtoRepr};
+use zksync_consensus_roles::{attester, validator};
+use zksync_protobuf::{read_required, required, ProtoFmt, ProtoRepr};
 use zksync_types::{
     abi, ethabi,
     fee::Fee,
@@ -20,6 +22,34 @@ use zksync_utils::{h256_to_u256, u256_to_h256};
 
 use crate::models::{parse_h160, parse_h256};
 
+/// Global attestation status served by
+/// `attestationStatus` RPC.
+#[derive(Debug, PartialEq, Clone)]
+pub struct AttestationStatus {
+    pub genesis: validator::GenesisHash,
+    pub next_batch_to_attest: attester::BatchNumber,
+}
+
+impl ProtoFmt for AttestationStatus {
+    type Proto = proto::AttestationStatus;
+
+    fn read(r: &Self::Proto) -> anyhow::Result<Self> {
+        Ok(Self {
+            genesis: read_required(&r.genesis).context("genesis")?,
+            next_batch_to_attest: attester::BatchNumber(
+                *required(&r.next_batch_to_attest).context("next_batch_to_attest")?,
+            ),
+        })
+    }
+
+    fn build(&self) -> Self::Proto {
+        Self::Proto {
+            genesis: Some(self.genesis.build()),
+            next_batch_to_attest: Some(self.next_batch_to_attest.0),
+        }
+    }
+}
+
 /// L2 block (= miniblock) payload.
 #[derive(Debug, PartialEq)]
 pub struct Payload {

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

@@ -2,6 +2,8 @@ syntax = "proto3";
 
 package zksync.dal;
 
+import "zksync/roles/validator.proto";
+
 message Payload {
   // zksync-era ProtocolVersionId
   optional uint32 protocol_version = 9; // required; u16
@@ -114,3 +116,8 @@ message PaymasterParams {
   optional bytes paymaster_address = 1; // required; H160
   optional bytes paymaster_input = 2; // required
 }
+
+message AttestationStatus {
+  optional roles.validator.GenesisHash genesis = 1; // required
+  optional uint64 next_batch_to_attest = 2; // required
+}

core/lib/dal/src/consensus/testonly.rs

@@ -0,0 +1,15 @@
+use rand::{
+    distributions::{Distribution, Standard},
+    Rng,
+};
+
+use super::AttestationStatus;
+
+impl Distribution<AttestationStatus> for Standard {
+    fn sample<R: Rng + ?Sized>(&self, rng: &mut R) -> AttestationStatus {
+        AttestationStatus {
+            genesis: rng.gen(),
+            next_batch_to_attest: rng.gen(),
+        }
+    }
+}

core/lib/dal/src/consensus/tests.rs

@@ -4,15 +4,15 @@ use rand::Rng;
 use zksync_concurrency::ctx;
 use zksync_protobuf::{
     repr::{decode, encode},
-    testonly::test_encode,
+    testonly::{test_encode, test_encode_random},
     ProtoRepr,
 };
 use zksync_test_account::Account;
 use zksync_types::{
     web3::Bytes, Execute, ExecuteTransactionCommon, L1BatchNumber, ProtocolVersionId, Transaction,
 };
 
-use super::{proto, Payload};
+use super::{proto, AttestationStatus, Payload};
 use crate::tests::mock_protocol_upgrade_transaction;
 
 fn execute(rng: &mut impl Rng) -> Execute {
@@ -59,6 +59,7 @@ fn payload(rng: &mut impl Rng, protocol_version: ProtocolVersionId) -> Payload {
 fn test_encoding() {
     let ctx = &ctx::test_root(&ctx::RealClock);
     let rng = &mut ctx.rng();
+    test_encode_random::<AttestationStatus>(rng);
     encode_decode::<proto::TransactionV25, ComparableTransaction>(l1_transaction(rng));
     encode_decode::<proto::TransactionV25, ComparableTransaction>(l2_transaction(rng));
     encode_decode::<proto::Transaction, ComparableTransaction>(l1_transaction(rng));

core/lib/dal/src/consensus_dal.rs

@@ -9,7 +9,7 @@ use zksync_db_connection::{
 };
 use zksync_types::L2BlockNumber;
 
-pub use crate::consensus::Payload;
+pub use crate::consensus::{AttestationStatus, Payload};
 use crate::{Core, CoreDal};
 
 /// Storage access methods for `zksync_core::consensus` module.
@@ -466,38 +466,82 @@ impl ConsensusDal<'_, '_> {
         )))
     }
 
-    /// Next batch that the attesters should vote for.
+    /// Number of L1 batch that the L2 block belongs to.
+    /// None if the L2 block doesn't exist.
+    async fn batch_of_block(
+        &mut self,
+        block: validator::BlockNumber,
+    ) -> anyhow::Result<Option<attester::BatchNumber>> {
+        let Some(row) = sqlx::query!(
+            r#"
+            SELECT
+                COALESCE(
+                    miniblocks.l1_batch_number,
+                    (
+                        SELECT
+                            (MAX(number) + 1)
+                        FROM
+                            l1_batches
+                    ),
+                    (
+                        SELECT
+                            MAX(l1_batch_number) + 1
+                        FROM
+                            snapshot_recovery
+                    )
+                ) AS "l1_batch_number!"
+            FROM
+                miniblocks
+            WHERE
+                number = $1
+            "#,
+            i64::try_from(block.0).context("overflow")?,
+        )
+        .instrument("batch_of_block")
+        .report_latency()
+        .fetch_optional(self.storage)
+        .await?
+        else {
+            return Ok(None);
+        };
+        Ok(Some(attester::BatchNumber(
+            row.l1_batch_number.try_into().context("overflow")?,
+        )))
+    }
+
+    /// Global attestation status.
+    /// Includes the next batch that the attesters should vote for.
+    /// None iff the consensus genesis is missing (i.e. consensus wasn't enabled) or
+    /// L2 block with number `genesis.first_block` doesn't exist yet.
+    ///
     /// This is a main node only query.
     /// ENs should call the attestation_status RPC of the main node.
-    pub async fn next_batch_to_attest(&mut self) -> anyhow::Result<attester::BatchNumber> {
-        // First batch that we don't have a certificate for.
-        if let Some(last) = self
-            .get_last_batch_certificate_number()
-            .await
-            .context("get_last_batch_certificate_number()")?
-        {
-            return Ok(last + 1);
-        }
-        // Otherwise start with the last sealed L1 batch.
-        // We don't want to backfill certificates for old batches.
-        // Note that there is a race condition in case the next
-        // batch is sealed before the certificate for the current
-        // last sealed batch is stored. This is only relevant
-        // for the first certificate though and anyway this is
-        // a test setup, so we are OK with that race condition.
-        if let Some(sealed) = self
-            .storage
-            .blocks_dal()
-            .get_sealed_l1_batch_number()
-            .await
-            .context("get_sealed_l1_batch_number()")?
-        {
-            return Ok(attester::BatchNumber(sealed.0.into()));
+    pub async fn attestation_status(&mut self) -> anyhow::Result<Option<AttestationStatus>> {
+        let Some(genesis) = self.genesis().await.context("genesis()")? else {
+            return Ok(None);
+        };
+        let Some(next_batch_to_attest) = async {
+            // First batch that we don't have a certificate for.
+            if let Some(last) = self
+                .get_last_batch_certificate_number()
+                .await
+                .context("get_last_batch_certificate_number()")?
+            {
+                return Ok(Some(last + 1));
+            }
+            // Otherwise start with the batch containing the first block of the fork.
+            self.batch_of_block(genesis.first_block)
+                .await
+                .context("batch_of_block()")
         }
-        // Otherwise start with 0.
-        // Note that main node doesn't start from snapshot
-        // and doesn't have prunning enabled.
-        Ok(attester::BatchNumber(0))
+        .await?
+        else {
+            return Ok(None);
+        };
+        Ok(Some(AttestationStatus {
+            genesis: genesis.hash(),
+            next_batch_to_attest,
+        }))
     }
 }
 

core/lib/types/src/api/en.rs

@@ -50,6 +50,4 @@ pub struct ConsensusGenesis(pub serde_json::Value);
 /// AttestationStatus maintained by the main node.
 /// Used for testing L1 batch signing by consensus attesters.
 #[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct AttestationStatus {
-    pub next_batch_to_attest: L1BatchNumber,
-}
+pub struct AttestationStatus(pub serde_json::Value);

core/lib/web3_decl/src/namespaces/en.rs

@@ -40,7 +40,7 @@ pub trait EnNamespace {
     /// This is a temporary RPC used for testing L1 batch signing
     /// by consensus attesters.
     #[method(name = "attestationStatus")]
-    async fn attestation_status(&self) -> RpcResult<en::AttestationStatus>;
+    async fn attestation_status(&self) -> RpcResult<Option<en::AttestationStatus>>;
 
     /// Get tokens that are white-listed and it can be used by paymasters.
     #[method(name = "whitelistedTokensForAA")]

core/node/api_server/src/web3/backend_jsonrpsee/namespaces/en.rs

@@ -25,7 +25,7 @@ impl EnNamespaceServer for EnNamespace {
             .map_err(|err| self.current_method().map_err(err))
     }
 
-    async fn attestation_status(&self) -> RpcResult<en::AttestationStatus> {
+    async fn attestation_status(&self) -> RpcResult<Option<en::AttestationStatus>> {
         self.attestation_status_impl()
             .await
             .map_err(|err| self.current_method().map_err(err))

core/node/api_server/src/web3/namespaces/en.rs

@@ -1,6 +1,5 @@
 use anyhow::Context as _;
 use zksync_config::{configs::EcosystemContracts, GenesisConfig};
-use zksync_consensus_roles::attester;
 use zksync_dal::{CoreDal, DalError};
 use zksync_types::{
     api::en, protocol_version::ProtocolSemanticVersion, tokens::TokenInfo, Address, L1BatchNumber,
@@ -17,12 +16,6 @@ pub(crate) struct EnNamespace {
     state: RpcState,
 }
 
-fn to_l1_batch_number(n: attester::BatchNumber) -> anyhow::Result<L1BatchNumber> {
-    Ok(L1BatchNumber(
-        n.0.try_into().context("L1BatchNumber overflow")?,
-    ))
-}
-
 impl EnNamespace {
     pub fn new(state: RpcState) -> Self {
         Self { state }
@@ -44,18 +37,30 @@ impl EnNamespace {
     }
 
     #[tracing::instrument(skip(self))]
-    pub async fn attestation_status_impl(&self) -> Result<en::AttestationStatus, Web3Error> {
-        Ok(en::AttestationStatus {
-            next_batch_to_attest: to_l1_batch_number(
-                self.state
-                    .acquire_connection()
-                    .await?
-                    .consensus_dal()
-                    .next_batch_to_attest()
-                    .await
-                    .context("next_batch_to_attest()")?,
-            )?,
-        })
+    pub async fn attestation_status_impl(
+        &self,
+    ) -> Result<Option<en::AttestationStatus>, Web3Error> {
+        let status = self
+            .state
+            .acquire_connection()
+            .await?
+            // unwrap is ok, because we start outermost transaction.
+            .transaction_builder()
+            .unwrap()
+            // run readonly transaction to perform consistent reads.
+            .set_readonly()
+            .build()
+            .await
+            .context("TransactionBuilder::build()")?
+            .consensus_dal()
+            .attestation_status()
+            .await?;
+
+        Ok(status.map(|s| {
+            en::AttestationStatus(
+                zksync_protobuf::serde::serialize(&s, serde_json::value::Serializer).unwrap(),
+            )
+        }))
     }
 
     pub(crate) fn current_method(&self) -> &MethodTracer {

core/node/consensus/src/testonly.rs

@@ -302,6 +302,11 @@ impl StateKeeper {
         validator::BlockNumber(self.last_block.0.into())
     }
 
+    /// Batch of the `last_block`.
+    pub fn last_batch(&self) -> L1BatchNumber {
+        self.last_batch
+    }
+
     /// Last L1 batch that has been sealed and will have
     /// metadata computed eventually.
     pub fn last_sealed_batch(&self) -> L1BatchNumber {