feat(en): Implement gossip fetcher

core/bin/external_node/src/main.rs

@@ -24,9 +24,8 @@ use zksync_core::{
     setup_sigint_handler,
     state_keeper::{L1BatchExecutorBuilder, MainBatchExecutorBuilder, ZkSyncStateKeeper},
     sync_layer::{
-        batch_status_updater::BatchStatusUpdater, external_io::ExternalIO,
-        fetcher::MainNodeFetcherCursor, genesis::perform_genesis_if_needed, ActionQueue,
-        MainNodeClient, SyncState,
+        batch_status_updater::BatchStatusUpdater, external_io::ExternalIO, fetcher::FetcherCursor,
+        genesis::perform_genesis_if_needed, ActionQueue, MainNodeClient, SyncState,
     },
 };
 use zksync_dal::{connection::DbVariant, healthcheck::ConnectionPoolHealthCheck, ConnectionPool};
@@ -129,7 +128,7 @@ async fn init_tasks(
             .await
             .context("failed to build a connection pool for `MainNodeFetcher`")?;
         let mut storage = pool.access_storage_tagged("sync_layer").await?;
-        MainNodeFetcherCursor::new(&mut storage)
+        FetcherCursor::new(&mut storage)
             .await
             .context("failed to load `MainNodeFetcher` cursor from Postgres")?
     };

core/lib/dal/migrations/20231102144901_add_consensus_fields_for_miniblocks.down.sql

@@ -0,0 +1,4 @@
+ALTER TABLE miniblocks
+    DROP COLUMN IF EXISTS commit_qc;
+ALTER TABLE miniblocks
+    DROP COLUMN IF EXISTS prev_consensus_block_hash;

core/lib/dal/migrations/20231102144901_add_consensus_fields_for_miniblocks.up.sql

@@ -0,0 +1,4 @@
+ALTER TABLE miniblocks
+    ADD COLUMN commit_qc BYTEA NULL;
+ALTER TABLE miniblocks
+    ADD COLUMN prev_consensus_block_hash BYTEA NULL;

core/lib/dal/sqlx-data.json

@@ -1344,6 +1344,110 @@
     },
     "query": "\n                UPDATE node_aggregation_witness_jobs_fri\n                SET status='queued'\n                WHERE (l1_batch_number, circuit_id, depth) IN\n                      (SELECT prover_jobs_fri.l1_batch_number, prover_jobs_fri.circuit_id, prover_jobs_fri.depth\n                       FROM prover_jobs_fri\n                                JOIN node_aggregation_witness_jobs_fri nawj ON\n                                prover_jobs_fri.l1_batch_number = nawj.l1_batch_number\n                                AND prover_jobs_fri.circuit_id = nawj.circuit_id\n                                AND prover_jobs_fri.depth = nawj.depth\n                       WHERE nawj.status = 'waiting_for_proofs'\n                         AND prover_jobs_fri.status = 'successful'\n                         AND prover_jobs_fri.aggregation_round = 1\n                         AND prover_jobs_fri.depth = 0\n                       GROUP BY prover_jobs_fri.l1_batch_number, prover_jobs_fri.circuit_id, prover_jobs_fri.depth, nawj.number_of_dependent_jobs\n                       HAVING COUNT(*) = nawj.number_of_dependent_jobs)\n                RETURNING l1_batch_number, circuit_id, depth;\n            "
   },
+  "1dc3019f127fd7aa760489457b6eba8dbbde21e03927e4ca71ebb6ab859ac8d1": {
+    "describe": {
+      "columns": [
+        {
+          "name": "number",
+          "ordinal": 0,
+          "type_info": "Int8"
+        },
+        {
+          "name": "l1_batch_number!",
+          "ordinal": 1,
+          "type_info": "Int8"
+        },
+        {
+          "name": "last_batch_miniblock?",
+          "ordinal": 2,
+          "type_info": "Int8"
+        },
+        {
+          "name": "timestamp",
+          "ordinal": 3,
+          "type_info": "Int8"
+        },
+        {
+          "name": "root_hash?",
+          "ordinal": 4,
+          "type_info": "Bytea"
+        },
+        {
+          "name": "l1_gas_price",
+          "ordinal": 5,
+          "type_info": "Int8"
+        },
+        {
+          "name": "l2_fair_gas_price",
+          "ordinal": 6,
+          "type_info": "Int8"
+        },
+        {
+          "name": "bootloader_code_hash",
+          "ordinal": 7,
+          "type_info": "Bytea"
+        },
+        {
+          "name": "default_aa_code_hash",
+          "ordinal": 8,
+          "type_info": "Bytea"
+        },
+        {
+          "name": "virtual_blocks",
+          "ordinal": 9,
+          "type_info": "Int8"
+        },
+        {
+          "name": "hash",
+          "ordinal": 10,
+          "type_info": "Bytea"
+        },
+        {
+          "name": "commit_qc",
+          "ordinal": 11,
+          "type_info": "Bytea"
+        },
+        {
+          "name": "prev_consensus_block_hash",
+          "ordinal": 12,
+          "type_info": "Bytea"
+        },
+        {
+          "name": "protocol_version!",
+          "ordinal": 13,
+          "type_info": "Int4"
+        },
+        {
+          "name": "fee_account_address?",
+          "ordinal": 14,
+          "type_info": "Bytea"
+        }
+      ],
+      "nullable": [
+        false,
+        null,
+        null,
+        false,
+        false,
+        false,
+        false,
+        true,
+        true,
+        false,
+        false,
+        true,
+        true,
+        true,
+        false
+      ],
+      "parameters": {
+        "Left": [
+          "Int8"
+        ]
+      }
+    },
+    "query": "SELECT miniblocks.number, COALESCE(miniblocks.l1_batch_number, (SELECT (max(number) + 1) FROM l1_batches)) as \"l1_batch_number!\", (SELECT max(m2.number) FROM miniblocks m2 WHERE miniblocks.l1_batch_number = m2.l1_batch_number) as \"last_batch_miniblock?\", miniblocks.timestamp, miniblocks.hash as \"root_hash?\", miniblocks.l1_gas_price, miniblocks.l2_fair_gas_price, miniblocks.bootloader_code_hash, miniblocks.default_aa_code_hash, miniblocks.virtual_blocks, miniblocks.hash, miniblocks.commit_qc, miniblocks.prev_consensus_block_hash, miniblocks.protocol_version as \"protocol_version!\", l1_batches.fee_account_address as \"fee_account_address?\" FROM miniblocks LEFT JOIN l1_batches ON miniblocks.l1_batch_number = l1_batches.number WHERE miniblocks.number = $1"
+  },
   "1ed353a16e8d0abaf426e5c235b20a79c727c08bc23fb1708a833a6930131691": {
     "describe": {
       "columns": [],
@@ -3887,98 +3991,6 @@
     },
     "query": "UPDATE eth_txs SET has_failed = TRUE WHERE id = $1"
   },
-  "5190fad25f0c476380af4013761d42ae97dbd55f87e38ceec33f8e148c5cbb14": {
-    "describe": {
-      "columns": [
-        {
-          "name": "number",
-          "ordinal": 0,
-          "type_info": "Int8"
-        },
-        {
-          "name": "l1_batch_number!",
-          "ordinal": 1,
-          "type_info": "Int8"
-        },
-        {
-          "name": "last_batch_miniblock?",
-          "ordinal": 2,
-          "type_info": "Int8"
-        },
-        {
-          "name": "timestamp",
-          "ordinal": 3,
-          "type_info": "Int8"
-        },
-        {
-          "name": "root_hash?",
-          "ordinal": 4,
-          "type_info": "Bytea"
-        },
-        {
-          "name": "l1_gas_price",
-          "ordinal": 5,
-          "type_info": "Int8"
-        },
-        {
-          "name": "l2_fair_gas_price",
-          "ordinal": 6,
-          "type_info": "Int8"
-        },
-        {
-          "name": "bootloader_code_hash",
-          "ordinal": 7,
-          "type_info": "Bytea"
-        },
-        {
-          "name": "default_aa_code_hash",
-          "ordinal": 8,
-          "type_info": "Bytea"
-        },
-        {
-          "name": "virtual_blocks",
-          "ordinal": 9,
-          "type_info": "Int8"
-        },
-        {
-          "name": "hash",
-          "ordinal": 10,
-          "type_info": "Bytea"
-        },
-        {
-          "name": "protocol_version!",
-          "ordinal": 11,
-          "type_info": "Int4"
-        },
-        {
-          "name": "fee_account_address?",
-          "ordinal": 12,
-          "type_info": "Bytea"
-        }
-      ],
-      "nullable": [
-        false,
-        null,
-        null,
-        false,
-        false,
-        false,
-        false,
-        true,
-        true,
-        false,
-        false,
-        true,
-        false
-      ],
-      "parameters": {
-        "Left": [
-          "Int8"
-        ]
-      }
-    },
-    "query": "\n                SELECT miniblocks.number,\n                    COALESCE(miniblocks.l1_batch_number, (SELECT (max(number) + 1) FROM l1_batches)) as \"l1_batch_number!\",\n                    (SELECT max(m2.number) FROM miniblocks m2 WHERE miniblocks.l1_batch_number = m2.l1_batch_number) as \"last_batch_miniblock?\",\n                    miniblocks.timestamp,\n                    miniblocks.hash as \"root_hash?\",\n                    miniblocks.l1_gas_price,\n                    miniblocks.l2_fair_gas_price,\n                    miniblocks.bootloader_code_hash,\n                    miniblocks.default_aa_code_hash,\n                    miniblocks.virtual_blocks,\n                    miniblocks.hash,\n                    miniblocks.protocol_version as \"protocol_version!\",\n                    l1_batches.fee_account_address as \"fee_account_address?\"\n                FROM miniblocks\n                LEFT JOIN l1_batches ON miniblocks.l1_batch_number = l1_batches.number\n                WHERE miniblocks.number = $1\n            "
-  },
   "51cb712685991ffd600dce59f5ed8b5a1bfce8feed46ebd02471c43802e6e65a": {
     "describe": {
       "columns": [
@@ -8733,6 +8745,20 @@
     },
     "query": "SELECT * FROM eth_txs_history WHERE eth_tx_id = $1 ORDER BY created_at DESC LIMIT 1"
   },
+  "ad48c26546edaa5f872e5698eb0b0d3ced291db178dd6c065fe6a39def4ea751": {
+    "describe": {
+      "columns": [],
+      "nullable": [],
+      "parameters": {
+        "Left": [
+          "Int8",
+          "Bytea",
+          "Bytea"
+        ]
+      }
+    },
+    "query": "UPDATE miniblocks SET prev_consensus_block_hash = $2, commit_qc = $3 WHERE number = $1"
+  },
   "ad4f74aa6f131df0243f4fa500ade1b98aa335bd71ed417b02361e2c697e60f8": {
     "describe": {
       "columns": [],

core/lib/dal/src/blocks_dal.rs

@@ -10,7 +10,7 @@ use sqlx::Row;
 
 use zksync_types::{
     aggregated_operations::AggregatedActionType,
-    block::{BlockGasCount, L1BatchHeader, MiniblockHeader},
+    block::{BlockGasCount, ConsensusBlockFields, L1BatchHeader, MiniblockHeader},
     commitment::{L1BatchMetadata, L1BatchWithMetadata},
     Address, L1BatchNumber, LogQuery, MiniblockNumber, ProtocolVersionId, H256,
     MAX_GAS_PER_PUBDATA_BYTE, U256,
@@ -466,6 +466,30 @@ impl BlocksDal<'_, '_> {
         Ok(())
     }
 
+    /// Sets consensus-related fields for the specified miniblock.
+    pub async fn set_miniblock_consensus_fields(
+        &mut self,
+        miniblock_number: MiniblockNumber,
+        consensus: &ConsensusBlockFields,
+    ) -> anyhow::Result<()> {
+        let result = sqlx::query!(
+            "UPDATE miniblocks \
+            SET prev_consensus_block_hash = $2, commit_qc = $3 \
+            WHERE number = $1",
+            miniblock_number.0 as i64,
+            consensus.prev_block_hash.as_bytes(),
+            &consensus.commit_qc_bytes.0
+        )
+        .execute(self.storage.conn())
+        .await?;
+
+        anyhow::ensure!(
+            result.rows_affected() == 1,
+            "Miniblock #{miniblock_number} is not present in Postgres"
+        );
+        Ok(())
+    }
+
     pub async fn update_hashes(
         &mut self,
         number_and_hashes: &[(MiniblockNumber, H256)],

core/lib/dal/src/models/storage_sync.rs

@@ -2,8 +2,9 @@ use std::convert::TryInto;
 
 use zksync_contracts::BaseSystemContractsHashes;
 use zksync_types::api::en::SyncBlock;
-use zksync_types::Transaction;
-use zksync_types::{Address, L1BatchNumber, MiniblockNumber, H256};
+use zksync_types::{
+    block::ConsensusBlockFields, Address, Bytes, L1BatchNumber, MiniblockNumber, Transaction, H256,
+};
 
 #[derive(Debug, Clone, sqlx::FromRow)]
 pub struct StorageSyncBlock {
@@ -22,6 +23,8 @@ pub struct StorageSyncBlock {
     pub protocol_version: i32,
     pub virtual_blocks: i64,
     pub hash: Vec<u8>,
+    pub commit_qc: Option<Vec<u8>>,
+    pub prev_consensus_block_hash: Option<Vec<u8>>,
 }
 
 impl StorageSyncBlock {
@@ -30,12 +33,15 @@ impl StorageSyncBlock {
         current_operator_address: Address,
         transactions: Option<Vec<Transaction>>,
     ) -> SyncBlock {
+        let number = self.number;
+        let commit_qc = self.commit_qc;
+
         SyncBlock {
             number: MiniblockNumber(self.number as u32),
             l1_batch_number: L1BatchNumber(self.l1_batch_number as u32),
             last_in_batch: self
                 .last_batch_miniblock
-                .map(|n| n == self.number)
+                .map(|n| n == number)
                 .unwrap_or(false),
             timestamp: self.timestamp as u64,
             root_hash: self.root_hash.as_deref().map(H256::from_slice),
@@ -60,6 +66,12 @@ impl StorageSyncBlock {
             virtual_blocks: Some(self.virtual_blocks as u32),
             hash: Some(H256::from_slice(&self.hash)),
             protocol_version: (self.protocol_version as u16).try_into().unwrap(),
+            consensus: self.prev_consensus_block_hash.and_then(|hash| {
+                Some(ConsensusBlockFields {
+                    prev_block_hash: H256::from_slice(&hash),
+                    commit_qc_bytes: Bytes(commit_qc?),
+                })
+            }),
         }
     }
 }

core/lib/dal/src/sync_dal.rs

@@ -23,24 +23,24 @@ impl SyncDal<'_, '_> {
         let latency = MethodLatency::new("sync_dal_sync_block");
         let storage_block_details = sqlx::query_as!(
             StorageSyncBlock,
-            r#"
-                SELECT miniblocks.number,
-                    COALESCE(miniblocks.l1_batch_number, (SELECT (max(number) + 1) FROM l1_batches)) as "l1_batch_number!",
-                    (SELECT max(m2.number) FROM miniblocks m2 WHERE miniblocks.l1_batch_number = m2.l1_batch_number) as "last_batch_miniblock?",
-                    miniblocks.timestamp,
-                    miniblocks.hash as "root_hash?",
-                    miniblocks.l1_gas_price,
-                    miniblocks.l2_fair_gas_price,
-                    miniblocks.bootloader_code_hash,
-                    miniblocks.default_aa_code_hash,
-                    miniblocks.virtual_blocks,
-                    miniblocks.hash,
-                    miniblocks.protocol_version as "protocol_version!",
-                    l1_batches.fee_account_address as "fee_account_address?"
-                FROM miniblocks
-                LEFT JOIN l1_batches ON miniblocks.l1_batch_number = l1_batches.number
-                WHERE miniblocks.number = $1
-            "#,
+            "SELECT miniblocks.number, \
+                COALESCE(miniblocks.l1_batch_number, (SELECT (max(number) + 1) FROM l1_batches)) as \"l1_batch_number!\", \
+                (SELECT max(m2.number) FROM miniblocks m2 WHERE miniblocks.l1_batch_number = m2.l1_batch_number) as \"last_batch_miniblock?\", \
+                miniblocks.timestamp, \
+                miniblocks.hash as \"root_hash?\", \
+                miniblocks.l1_gas_price, \
+                miniblocks.l2_fair_gas_price, \
+                miniblocks.bootloader_code_hash, \
+                miniblocks.default_aa_code_hash, \
+                miniblocks.virtual_blocks, \
+                miniblocks.hash, \
+                miniblocks.commit_qc, \
+                miniblocks.prev_consensus_block_hash, \
+                miniblocks.protocol_version as \"protocol_version!\", \
+                l1_batches.fee_account_address as \"fee_account_address?\" \
+            FROM miniblocks \
+            LEFT JOIN l1_batches ON miniblocks.l1_batch_number = l1_batches.number \
+            WHERE miniblocks.number = $1",
             block_number.0 as i64
         )
         .instrument("sync_dal_sync_block.block")