Merge pull request #30 from Blobscan/feature/parallel-indexing

Add Parallel Indexing

Author: PJColombo

.env.example

@@ -4,3 +4,4 @@ SECRET_KEY=supersecret
 BLOBSCAN_API_ENDPOINT=
 EXECUTION_NODE_URL=
 BEACON_NODE_RPC=
+NUM_PROCESSING_THREADS=

src/beacon_client/mod.rs

@@ -2,7 +2,7 @@ use reqwest::{Client, StatusCode};
 use std::time::Duration;
 
 use self::types::{
-    BeaconClientError, BeaconClientResult, BlobData, BlobsResponse, BlockMessage as Block,
+    BeaconClientError, BeaconClientResult, Blob, BlobsResponse, BlockMessage as Block,
     BlockResponse,
 };
 
@@ -11,7 +11,7 @@ pub mod types;
 #[derive(Debug, Clone)]
 pub struct BeaconClient {
     base_url: String,
-    client: reqwest::Client,
+    client: Client,
 }
 
 pub struct Config {
@@ -20,25 +20,18 @@ pub struct Config {
 }
 
 impl BeaconClient {
-    pub fn try_from(config: Config) -> BeaconClientResult<Self> {
-        let mut client_builder = Client::builder();
-
-        if let Some(timeout) = config.timeout {
-            client_builder = client_builder.timeout(timeout);
-        }
-
-        Ok(Self {
+    pub fn with_client(client: Client, config: Config) -> Self {
+        Self {
             base_url: config.base_url,
-            client: client_builder.build()?,
-        })
+            client,
+        }
     }
 
     pub async fn get_block(&self, slot: Option<u32>) -> BeaconClientResult<Option<Block>> {
         let slot = match slot {
             Some(slot) => slot.to_string(),
             None => String::from("head"),
         };
-
         let url = self.build_url(&format!("eth/v2/beacon/blocks/{slot}"));
 
         let block_response = self.client.get(url).send().await?;
@@ -54,7 +47,7 @@ impl BeaconClient {
         }
     }
 
-    pub async fn get_blobs(&self, slot: u32) -> BeaconClientResult<Option<Vec<BlobData>>> {
+    pub async fn get_blobs(&self, slot: u32) -> BeaconClientResult<Option<Vec<Blob>>> {
         let url = self.build_url(&format!("eth/v1/beacon/blobs/{slot}"));
 
         let blobs_response = self.client.get(url).send().await?;

src/beacon_client/types.rs

@@ -1,43 +1,43 @@
 use anyhow::Result;
 use ethers::types::{Bytes, H256};
-use serde::{Deserialize, Serialize};
+use serde::Deserialize;
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Deserialize, Debug)]
 pub struct ExecutionPayload {
     pub block_hash: H256,
 }
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Deserialize, Debug)]
 pub struct BlockBody {
     pub execution_payload: Option<ExecutionPayload>,
     pub blob_kzg_commitments: Option<Vec<String>>,
 }
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Deserialize, Debug)]
 pub struct BlockMessage {
     pub slot: String,
     pub body: BlockBody,
 }
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Deserialize, Debug)]
 pub struct Block {
     pub message: BlockMessage,
 }
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Deserialize, Debug)]
 pub struct BlockResponse {
     pub data: Block,
 }
 
-#[derive(Serialize, Deserialize, Debug)]
-pub struct BlobData {
+#[derive(Deserialize, Debug)]
+pub struct Blob {
     pub index: String,
     pub kzg_commitment: String,
     pub blob: Bytes,
 }
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Deserialize, Debug)]
 pub struct BlobsResponse {
-    pub data: Vec<BlobData>,
+    pub data: Vec<Blob>,
 }
 
 #[derive(Debug, thiserror::Error)]

src/blobscan_client/jwt_manager.rs

@@ -35,7 +35,7 @@ impl JWTManager {
             refresh_interval: config.refresh_interval,
             safety_margin: match config.safety_magin {
                 Some(safety_margin) => safety_margin,
-                None => Duration::seconds(10),
+                None => Duration::minutes(1),
             },
         }
     }

src/blobscan_client/mod.rs

@@ -5,16 +5,16 @@ use reqwest::{Client, StatusCode};
 use self::{
     jwt_manager::{Config as JWTManagerConfig, JWTManager},
     types::{
-        BlobEntity, BlobscanClientError, BlobscanClientResult, BlockEntity, IndexRequest,
-        SlotRequest, SlotResponse, TransactionEntity,
+        Blob, BlobscanClientError, BlobscanClientResult, Block, IndexRequest, SlotRequest,
+        SlotResponse, Transaction,
     },
 };
 
 mod jwt_manager;
 
 pub mod types;
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct BlobscanClient {
     base_url: String,
     client: reqwest::Client,
@@ -27,30 +27,28 @@ pub struct Config {
     pub timeout: Option<Duration>,
 }
 
-impl BlobscanClient {
-    pub fn try_from(config: Config) -> BlobscanClientResult<Self> {
-        let mut client_builder = Client::builder();
-
-        if let Some(timeout) = config.timeout {
-            client_builder = client_builder.timeout(timeout);
-        }
+pub fn build_jwt_manager(secret_key: String) -> JWTManager {
+    JWTManager::new(JWTManagerConfig {
+        secret_key,
+        refresh_interval: chrono::Duration::hours(1),
+        safety_magin: None,
+    })
+}
 
-        Ok(Self {
+impl BlobscanClient {
+    pub fn with_client(client: Client, config: Config) -> Self {
+        Self {
             base_url: config.base_url,
-            client: client_builder.build()?,
-            jwt_manager: JWTManager::new(JWTManagerConfig {
-                secret_key: config.secret_key,
-                refresh_interval: chrono::Duration::minutes(30),
-                safety_magin: None,
-            }),
-        })
+            client,
+            jwt_manager: build_jwt_manager(config.secret_key),
+        }
     }
 
     pub async fn index(
         &self,
-        block: BlockEntity,
-        transactions: Vec<TransactionEntity>,
-        blobs: Vec<BlobEntity>,
+        block: Block,
+        transactions: Vec<Transaction>,
+        blobs: Vec<Blob>,
     ) -> BlobscanClientResult<()> {
         let path = String::from("index");
         let url = self.build_url(&path);
@@ -63,17 +61,15 @@ impl BlobscanClient {
 
         let index_response = self
             .client
-            .post(url)
+            .put(url)
             .bearer_auth(token)
             .json(&index_request)
             .send()
             .await?;
 
         match index_response.status() {
             StatusCode::OK => Ok(()),
-            _ => Err(BlobscanClientError::BlobscanClientError(
-                index_response.text().await?,
-            )),
+            _ => Err(BlobscanClientError::ApiError(index_response.text().await?)),
         }
     }
 
@@ -92,9 +88,7 @@ impl BlobscanClient {
 
         match slot_response.status() {
             StatusCode::OK => Ok(()),
-            _ => Err(BlobscanClientError::BlobscanClientError(
-                slot_response.text().await?,
-            )),
+            _ => Err(BlobscanClientError::ApiError(slot_response.text().await?)),
         }
     }
 
@@ -107,9 +101,7 @@ impl BlobscanClient {
         match slot_response.status() {
             StatusCode::OK => Ok(Some(slot_response.json::<SlotResponse>().await?.slot)),
             StatusCode::NOT_FOUND => Ok(None),
-            _ => Err(BlobscanClientError::BlobscanClientError(
-                slot_response.text().await?,
-            )),
+            _ => Err(BlobscanClientError::ApiError(slot_response.text().await?)),
         }
     }
 

src/blobscan_client/types.rs

@@ -4,68 +4,86 @@ use ethers::types::{
 };
 use serde::{Deserialize, Serialize};
 
-use crate::{beacon_client::types::BlobData, utils::web3::calculate_versioned_hash};
+use crate::{beacon_client::types::Blob as BeaconBlob, utils::web3::calculate_versioned_hash};
 
 #[derive(Serialize, Deserialize, Debug)]
-pub struct BlockEntity {
+pub struct Block {
     pub number: U64,
     pub hash: H256,
     pub timestamp: U256,
     pub slot: u32,
 }
 
 #[derive(Serialize, Deserialize, Debug)]
-pub struct TransactionEntity {
+#[serde(rename_all = "camelCase")]
+pub struct Transaction {
     pub hash: H256,
     pub from: Address,
-    pub to: Address,
-    #[serde(rename = "blockNumber")]
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub to: Option<Address>,
     pub block_number: U64,
 }
 
 #[derive(Serialize, Deserialize, Debug)]
-pub struct BlobEntity {
-    #[serde(rename = "versionedHash")]
+#[serde(rename_all = "camelCase")]
+pub struct Blob {
     pub versioned_hash: H256,
     pub commitment: String,
     pub data: Bytes,
-    #[serde(rename = "txHash")]
     pub tx_hash: H256,
     pub index: u32,
 }
 
 #[derive(Serialize, Deserialize, Debug)]
-pub struct SlotResponse {
-    pub slot: u32,
+#[serde(rename_all = "camelCase")]
+pub struct FailedSlotsChunk {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub id: Option<u32>,
+    pub initial_slot: u32,
+    pub final_slot: u32,
 }
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Debug)]
 pub struct SlotRequest {
     pub slot: u32,
 }
+#[derive(Deserialize, Debug)]
+pub struct SlotResponse {
+    pub slot: u32,
+}
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Debug)]
 pub struct IndexRequest {
-    pub block: BlockEntity,
-    pub transactions: Vec<TransactionEntity>,
-    pub blobs: Vec<BlobEntity>,
+    pub block: Block,
+    pub transactions: Vec<Transaction>,
+    pub blobs: Vec<Blob>,
 }
 
 #[derive(Debug, thiserror::Error)]
 pub enum BlobscanClientError {
     #[error(transparent)]
     Reqwest(#[from] reqwest::Error),
 
-    #[error("Blobscan client error: {0}")]
-    BlobscanClientError(String),
+    #[error("API usage error: {0}")]
+    ApiError(String),
 
     #[error(transparent)]
-    JWTError(#[from] anyhow::Error),
+    Other(#[from] anyhow::Error),
 }
 
 pub type BlobscanClientResult<T> = Result<T, BlobscanClientError>;
 
-impl<'a> TryFrom<(&'a EthersBlock<EthersTransaction>, u32)> for BlockEntity {
+impl From<(u32, u32)> for FailedSlotsChunk {
+    fn from((initial_slot, final_slot): (u32, u32)) -> Self {
+        Self {
+            id: None,
+            initial_slot,
+            final_slot,
+        }
+    }
+}
+
+impl<'a> TryFrom<(&'a EthersBlock<EthersTransaction>, u32)> for Block {
     type Error = anyhow::Error;
 
     fn try_from(
@@ -86,9 +104,7 @@ impl<'a> TryFrom<(&'a EthersBlock<EthersTransaction>, u32)> for BlockEntity {
     }
 }
 
-impl<'a> TryFrom<(&'a EthersTransaction, &'a EthersBlock<EthersTransaction>)>
-    for TransactionEntity
-{
+impl<'a> TryFrom<(&'a EthersTransaction, &'a EthersBlock<EthersTransaction>)> for Transaction {
     type Error = anyhow::Error;
 
     fn try_from(
@@ -102,18 +118,16 @@ impl<'a> TryFrom<(&'a EthersTransaction, &'a EthersBlock<EthersTransaction>)>
                 .with_context(|| "Missing block number field in execution block".to_string())?,
             hash,
             from: ethers_tx.from,
-            to: ethers_tx
-                .to
-                .with_context(|| format!("Missing to field in transaction {hash}"))?,
+            to: ethers_tx.to,
         })
     }
 }
 
-impl<'a> TryFrom<(&'a BlobData, u32, H256)> for BlobEntity {
+impl<'a> TryFrom<(&'a BeaconBlob, u32, H256)> for Blob {
     type Error = anyhow::Error;
 
     fn try_from(
-        (blob_data, index, tx_hash): (&'a BlobData, u32, H256),
+        (blob_data, index, tx_hash): (&'a BeaconBlob, u32, H256),
     ) -> Result<Self, Self::Error> {
         Ok(Self {
             tx_hash,
@@ -125,9 +139,9 @@ impl<'a> TryFrom<(&'a BlobData, u32, H256)> for BlobEntity {
     }
 }
 
-impl<'a> From<(&'a BlobData, &'a H256, usize, &'a H256)> for BlobEntity {
+impl<'a> From<(&'a BeaconBlob, &'a H256, usize, &'a H256)> for Blob {
     fn from(
-        (blob_data, versioned_hash, index, tx_hash): (&'a BlobData, &'a H256, usize, &'a H256),
+        (blob_data, versioned_hash, index, tx_hash): (&'a BeaconBlob, &'a H256, usize, &'a H256),
     ) -> Self {
         Self {
             tx_hash: *tx_hash,