Merge pull request #122 from bitfinity-network/add_certification

Add lest certified block data to block extractor

Author: ufoscout

src/evm-block-extractor/Cargo.toml

@@ -17,6 +17,7 @@ ethers-core = { workspace = true }
 futures = { workspace = true }
 gcp-bigquery-client = { workspace = true }
 hex = { workspace = true }
+jsonrpc-core = { workspace = true }
 jsonrpsee = { workspace = true }
 lightspeed_scheduler = { workspace = true }
 log = { workspace = true }

src/evm-block-extractor/src/database/big_query_db_client.rs

@@ -17,9 +17,13 @@ use serde::de::DeserializeOwned;
 use serde::Serialize;
 use serde_json::Value;
 
-use super::{AccountBalance, DataContainer, DatabaseClient, CHAIN_ID_KEY, GENESIS_BALANCES_KEY};
+use super::{
+    AccountBalance, CertifiedBlock, DataContainer, DatabaseClient, CHAIN_ID_KEY,
+    GENESIS_BALANCES_KEY,
+};
 
 const BQ_BLOCKS_TABLE_ID: &str = "blocks";
+const BQ_CERTIFIED_BLOCKS_TABLE_ID: &str = "certified_blocks";
 const BQ_TRANSACTIONS_TABLE_ID: &str = "transactions";
 const BQ_KEY_VALUE_TABLE_ID: &str = "key_value_data";
 
@@ -139,6 +143,13 @@ impl BigQueryDbClient {
                     TableFieldSchema::json("body"),
                 ],
             ),
+            (
+                BQ_CERTIFIED_BLOCKS_TABLE_ID,
+                vec![
+                    TableFieldSchema::integer("id"),
+                    TableFieldSchema::json("certified_response"),
+                ],
+            ),
             (
                 BQ_TRANSACTIONS_TABLE_ID,
                 vec![
@@ -481,6 +492,36 @@ impl DatabaseClient for BigQueryDbClient {
             .await
     }
 
+    async fn insert_certified_block_data(&self, block: CertifiedBlock) -> anyhow::Result<()> {
+        let hash = block.data.hash.to_hex_str();
+        let block_row = CertifiedBlockRow {
+            id: block.data.number.0.as_u64(),
+            certified_response: serde_json::to_value(block).expect("Failed to serialize block"),
+        };
+
+        let rows = TableDataInsertAllRequestRows {
+            insert_id: Some(hash),
+            json: serde_json::to_value(block_row).expect("Failed to serialize block"),
+        };
+
+        self.insert_batch_data(BQ_CERTIFIED_BLOCKS_TABLE_ID, vec![rows])
+            .await
+    }
+
+    async fn get_last_certified_block_data(&self) -> anyhow::Result<CertifiedBlock> {
+        let query_request = QueryRequest {
+            query_parameters: None,
+            query: format!(
+                "SELECT certified_response FROM `{project_id}.{dataset_id}.{table_id}` ORDER BY id DESC LIMIT 1",
+                project_id = self.project_id,
+                dataset_id = self.dataset_id,
+                table_id = BQ_CERTIFIED_BLOCKS_TABLE_ID,),
+                ..Default::default()
+            };
+
+        self.query_one(query_request).await
+    }
+
     async fn get_transaction(&self, tx_hash: H256) -> anyhow::Result<Transaction> {
         let query_request = QueryRequest {
             query_parameters: Some(vec![
@@ -517,6 +558,13 @@ struct BlockRow {
     body: Value,
 }
 
+#[derive(Debug, Serialize)]
+pub struct CertifiedBlockRow {
+    id: u64,
+    #[serde(serialize_with = "serialize_json_as_string")]
+    certified_response: Value,
+}
+
 #[derive(Debug, Serialize, Clone)]
 struct ExeResultRow {
     tx_hash: String,

src/evm-block-extractor/src/database/mod.rs

@@ -1,6 +1,7 @@
 pub mod big_query_db_client;
 pub mod postgres_db_client;
 
+use did::certified::CertifiedResult;
 use did::{Block, Transaction, H160, H256, U256};
 use serde::{Deserialize, Serialize};
 
@@ -28,6 +29,9 @@ const GENESIS_BALANCES_KEY: &str = "genesis_balances";
 /// The chain id key in the key value store
 const CHAIN_ID_KEY: &str = "chain_id";
 
+/// Certified block data
+pub type CertifiedBlock = CertifiedResult<Block<H256>>;
+
 /// A trait for interacting with a blockchain database
 #[async_trait::async_trait]
 pub trait DatabaseClient: Send + Sync {
@@ -60,6 +64,12 @@ pub trait DatabaseClient: Send + Sync {
         transactions: &[Transaction],
     ) -> anyhow::Result<()>;
 
+    /// Insert certified block data
+    async fn insert_certified_block_data(&self, response: CertifiedBlock) -> anyhow::Result<()>;
+
+    /// Returns certified response for the last block
+    async fn get_last_certified_block_data(&self) -> anyhow::Result<CertifiedBlock>;
+
     /// Get genesis balances
     async fn get_genesis_balances(&self) -> anyhow::Result<Option<Vec<AccountBalance>>>;
 

src/evm-block-extractor/src/database/postgres_db_client.rs

@@ -5,7 +5,10 @@ use serde::de::DeserializeOwned;
 use serde::Serialize;
 use sqlx::postgres::PgRow;
 
-use super::{AccountBalance, DataContainer, DatabaseClient, CHAIN_ID_KEY, GENESIS_BALANCES_KEY};
+use super::{
+    AccountBalance, CertifiedBlock, DataContainer, DatabaseClient, CHAIN_ID_KEY,
+    GENESIS_BALANCES_KEY,
+};
 
 static MIGRATOR: Migrator = ::sqlx::migrate!("src_resources/db/postgres/migrations");
 
@@ -150,6 +153,30 @@ impl DatabaseClient for PostgresDbClient {
         Ok(())
     }
 
+    async fn insert_certified_block_data(&self, response: CertifiedBlock) -> anyhow::Result<()> {
+        let block_id = response.data.number.0.as_u64();
+
+        let mut tx = self.pool.begin().await?;
+        sqlx::query("INSERT INTO CERTIFIED_EVM_BLOCK (id, certified_response) VALUES ($1, $2) ON CONFLICT (id) DO UPDATE SET certified_response = $2")
+                    .bind(block_id as i64)
+                    .bind(serde_json::to_value(response)?)
+                    .execute(&mut *tx)
+                    .await
+                    .map_err(|e| anyhow::anyhow!("Error inserting certified block {}: {:?}", block_id, e))
+                    .map(|_| ())?;
+        tx.commit().await?;
+
+        Ok(())
+    }
+
+    async fn get_last_certified_block_data(&self) -> anyhow::Result<CertifiedBlock> {
+        sqlx::query("SELECT certified_response FROM CERTIFIED_EVM_BLOCK ORDER BY id DESC LIMIT 1")
+            .fetch_one(&self.pool)
+            .await
+            .map_err(|e| anyhow::anyhow!("Error getting last certified block: {:?}", e))
+            .and_then(|row| from_row_value(&row, 0))
+    }
+
     /// Get the latest block number
     async fn get_latest_block_number(&self) -> anyhow::Result<Option<u64>> {
         sqlx::query("SELECT MAX(id) FROM EVM_BLOCK")

src/evm-block-extractor/src/rpc.rs

@@ -5,7 +5,7 @@ use ethers_core::utils::rlp::{RlpStream, EMPTY_LIST_RLP};
 use jsonrpsee::core::RpcResult;
 use jsonrpsee::proc_macros::rpc;
 
-use crate::database::DatabaseClient;
+use crate::database::{CertifiedBlock, DatabaseClient};
 
 #[derive(Clone)]
 pub struct EthImpl {
@@ -46,6 +46,9 @@ pub trait IC {
 
     #[method(name = "getGenesisBalances")]
     async fn get_genesis_balances(&self) -> RpcResult<Vec<(H160, U256)>>;
+
+    #[method(name = "getLastBlockCertifiedData")]
+    async fn get_last_block_certified_data(&self) -> RpcResult<CertifiedBlock>;
 }
 
 #[async_trait::async_trait]
@@ -114,6 +117,19 @@ impl ICServer for EthImpl {
             .map(|account| (account.address.into(), account.balance.into()))
             .collect())
     }
+
+    async fn get_last_block_certified_data(&self) -> RpcResult<CertifiedBlock> {
+        let certified_data = self
+            .blockchain
+            .get_last_certified_block_data()
+            .await
+            .map_err(|e| {
+                log::error!("Error getting last block certified data: {:?}", e);
+                jsonrpsee::types::error::ErrorCode::InternalError
+            })?;
+
+        Ok(certified_data)
+    }
 }
 
 #[async_trait::async_trait]

src/evm-block-extractor/src/task/block_extractor.rs

@@ -7,7 +7,7 @@ use log::*;
 use tokio::time::Duration;
 
 use crate::config::ExtractorArgs;
-use crate::database::{AccountBalance, DatabaseClient};
+use crate::database::{AccountBalance, CertifiedBlock, DatabaseClient};
 
 /// Starts the block extractor process
 pub async fn start_extractor(
@@ -37,7 +37,7 @@ pub async fn start_extractor(
     debug!("latest block number stored: {:?}", start_block);
 
     extractor
-        .collect_blocks(start_block.map(|b| b + 1).unwrap_or_default(), end_block)
+        .collect_all(start_block.map(|b| b + 1).unwrap_or_default(), end_block)
         .await?;
 
     Ok(())
@@ -69,13 +69,14 @@ impl BlockExtractor {
     /// Collects blocks from the EVMC and stores them in the database.
     /// Returns the inclusive range of blocks that were collected.
     /// This collects also the genesis accounts if needed.
-    pub async fn collect_blocks(
+    pub async fn collect_all(
         &mut self,
         from_block_inclusive: u64,
         to_block_inclusive: u64,
     ) -> anyhow::Result<(u64, u64)> {
         self.collect_chain_id().await?;
         self.collect_genesis_balances().await?;
+        self.collect_last_certified_block().await?;
 
         info!(
             "Getting blocks from {:?} to {}",
@@ -132,6 +133,25 @@ impl BlockExtractor {
         Ok((from_block_inclusive, to_block_inclusive))
     }
 
+    /// Collects last certified block
+    async fn collect_last_certified_block(&self) -> anyhow::Result<()> {
+        const JSON_RPC_METHOD_LAST_CERTIFIED_BLOCK: &str = "ic_getLastCertifiedBlock";
+
+        let certified_block = self
+            .client
+            .single_request::<CertifiedBlock>(
+                JSON_RPC_METHOD_LAST_CERTIFIED_BLOCK.to_string(),
+                jsonrpc_core::Params::Array(vec![]),
+                jsonrpc_core::Id::Null,
+            )
+            .await?;
+        self.blockchain
+            .insert_certified_block_data(certified_block)
+            .await?;
+
+        Ok(())
+    }
+
     /// Collects the genesis accounts if needed.
     async fn collect_genesis_balances(&self) -> anyhow::Result<()> {
         if self.blockchain.get_genesis_balances().await?.is_some() {

src/evm-block-extractor/src_resources/db/postgres/migrations/00001_create_schema.sql

@@ -10,7 +10,6 @@ create table EVM_BLOCK (
 
 -- End - EVM_BLOCK -
 
-
 -----------------------------------------
 -- Begin - EVM_TRANSACTION_EXE_RESULT -
 -----------------------------------------

src/evm-block-extractor/src_resources/db/postgres/migrations/00003_add_certified_blocks.sql

@@ -0,0 +1,10 @@
+-----------------------------
+-- Begin - CERTIFIED_EVM_BLOCK -
+-----------------------------
+
+create table CERTIFIED_EVM_BLOCK (
+    ID bigint primary key,
+    CERTIFIED_RESPONSE JSONB
+);
+
+-- End - CERTIFIED_EVM_BLOCK -

src/evm-block-extractor/tests/tests/block_extractor_it.rs

@@ -29,10 +29,7 @@ async fn test_extractor_collect_blocks() {
 
         println!("Getting blocks from {:?} to {}", start_block, end_block);
 
-        let result = extractor
-            .collect_blocks(start_block, end_block)
-            .await
-            .unwrap();
+        let result = extractor.collect_all(start_block, end_block).await.unwrap();
 
         assert_eq!(result.0, start_block);
         assert_eq!(result.1, end_block);
@@ -63,6 +60,17 @@ async fn test_extractor_collect_blocks() {
             assert_eq!(evmc_chain_id, db_chain_id);
         }
 
+        // Check last certified block
+        {
+            let certified_data = db_client.get_last_certified_block_data().await.unwrap();
+            assert!(!certified_data.certificate.is_empty());
+            assert!(!certified_data.witness.is_empty());
+
+            // Check that it is more or less last block
+            assert!(end_block - 10 <= certified_data.data.number.0.as_u64());
+            assert!(end_block + 10 >= certified_data.data.number.0.as_u64());
+        }
+
         for block_num in start_block..=end_block {
             let block = db_client.get_block_by_number(block_num).await.unwrap();
 

src/evm-block-extractor/tests/tests/database_client_it.rs

@@ -1,5 +1,5 @@
 use did::{Block, Transaction, H160, H256, U256, U64};
-use evm_block_extractor::database::AccountBalance;
+use evm_block_extractor::database::{AccountBalance, CertifiedBlock};
 use rand::random;
 
 use crate::test_with_clients;
@@ -533,3 +533,50 @@ async fn test_insert_and_fetch_chain_id() {
     })
     .await;
 }
+
+#[tokio::test]
+async fn test_insert_and_fetch_last_block_certified_data() {
+    test_with_clients(|db_client| async move {
+        // Arrange
+        db_client.init(None, false).await.unwrap();
+
+        let block_1 = Block::<H256> {
+            number: 1u64.into(),
+            ..Default::default()
+        };
+        let block_2 = Block::<H256> {
+            number: 2u64.into(),
+            ..Default::default()
+        };
+
+        let result_1 = CertifiedBlock {
+            certificate: vec![1, 2, 3],
+            witness: vec![5, 6, 7],
+            data: block_1,
+        };
+        let result_2 = CertifiedBlock {
+            certificate: vec![11, 12, 13],
+            witness: vec![15, 16, 17],
+            data: block_2,
+        };
+
+        db_client
+            .insert_certified_block_data(result_1.clone())
+            .await
+            .unwrap();
+        assert_eq!(
+            result_1,
+            db_client.get_last_certified_block_data().await.unwrap()
+        );
+
+        db_client
+            .insert_certified_block_data(result_2.clone())
+            .await
+            .unwrap();
+        assert_eq!(
+            result_2,
+            db_client.get_last_certified_block_data().await.unwrap()
+        );
+    })
+    .await;
+}