Merge pull request #81 from Blobscan/fix/reorg-handling

fix: improve reorg handling logic

Author: PJColombo

src/clients/beacon/mod.rs

@@ -9,13 +9,14 @@ use reqwest_eventsource::EventSource;
 
 #[cfg(test)]
 use mockall::automock;
+use types::BlockHeader;
 
 use crate::{
     clients::{beacon::types::BlockHeaderResponse, common::ClientResult},
     json_get,
 };
 
-use self::types::{Blob, BlobsResponse, Block, BlockHeader, BlockId, BlockResponse, Topic};
+use self::types::{Blob, BlobsResponse, Block, BlockId, BlockResponse, Topic};
 
 pub mod types;
 
@@ -34,9 +35,9 @@ pub struct Config {
 #[async_trait]
 #[cfg_attr(test, automock)]
 pub trait CommonBeaconClient: Send + Sync + Debug {
-    async fn get_block(&self, block_id: &BlockId) -> ClientResult<Option<Block>>;
-    async fn get_block_header(&self, block_id: &BlockId) -> ClientResult<Option<BlockHeader>>;
-    async fn get_blobs(&self, block_id: &BlockId) -> ClientResult<Option<Vec<Blob>>>;
+    async fn get_block(&self, block_id: BlockId) -> ClientResult<Option<Block>>;
+    async fn get_block_header(&self, block_id: BlockId) -> ClientResult<Option<BlockHeader>>;
+    async fn get_blobs(&self, block_id: BlockId) -> ClientResult<Option<Vec<Blob>>>;
     fn subscribe_to_events(&self, topics: &[Topic]) -> ClientResult<EventSource>;
 }
 
@@ -56,7 +57,7 @@ impl BeaconClient {
 
 #[async_trait]
 impl CommonBeaconClient for BeaconClient {
-    async fn get_block(&self, block_id: &BlockId) -> ClientResult<Option<Block>> {
+    async fn get_block(&self, block_id: BlockId) -> ClientResult<Option<Block>> {
         let path = format!("v2/beacon/blocks/{}", { block_id.to_detailed_string() });
         let url = self.base_url.join(path.as_str())?;
 
@@ -66,7 +67,7 @@ impl CommonBeaconClient for BeaconClient {
         })
     }
 
-    async fn get_block_header(&self, block_id: &BlockId) -> ClientResult<Option<BlockHeader>> {
+    async fn get_block_header(&self, block_id: BlockId) -> ClientResult<Option<BlockHeader>> {
         let path = format!("v1/beacon/headers/{}", { block_id.to_detailed_string() });
         let url = self.base_url.join(path.as_str())?;
 
@@ -77,12 +78,12 @@ impl CommonBeaconClient for BeaconClient {
             self.exp_backoff.clone()
         )
         .map(|res| match res {
-            Some(r) => Some(r.data),
+            Some(r) => Some(r.into()),
             None => None,
         })
     }
 
-    async fn get_blobs(&self, block_id: &BlockId) -> ClientResult<Option<Vec<Blob>>> {
+    async fn get_blobs(&self, block_id: BlockId) -> ClientResult<Option<Vec<Blob>>> {
         let path = format!("v1/beacon/blob_sidecars/{}", {
             block_id.to_detailed_string()
         });

src/clients/beacon/types.rs

@@ -3,6 +3,10 @@ use std::{fmt, str::FromStr};
 use alloy::primitives::{Bytes, B256};
 use serde::{Deserialize, Serialize};
 
+use crate::clients::common::ClientError;
+
+use super::CommonBeaconClient;
+
 #[derive(Serialize, Debug, Clone, PartialEq)]
 pub enum BlockId {
     Head,
@@ -33,6 +37,9 @@ pub struct BlockBody {
 #[derive(Deserialize, Debug)]
 pub struct BlockMessage {
     pub body: BlockBody,
+    pub parent_root: B256,
+    #[serde(deserialize_with = "deserialize_number")]
+    pub slot: u32,
 }
 
 #[derive(Deserialize, Debug)]
@@ -59,11 +66,18 @@ pub struct BlobsResponse {
 
 #[derive(Deserialize, Debug)]
 pub struct BlockHeaderResponse {
-    pub data: BlockHeader,
+    pub data: BlockHeaderData,
 }
 
-#[derive(Deserialize, Debug)]
+#[derive(Deserialize, Debug, Clone)]
 pub struct BlockHeader {
+    pub root: B256,
+    pub parent_root: B256,
+    pub slot: u32,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct BlockHeaderData {
     pub root: B256,
     pub header: InnerBlockHeader,
 }
@@ -83,6 +97,7 @@ pub struct BlockHeaderMessage {
 pub struct HeadEventData {
     #[serde(deserialize_with = "deserialize_number")]
     pub slot: u32,
+    #[allow(dead_code)]
     pub block: B256,
 }
 
@@ -156,3 +171,65 @@ impl From<&Topic> for String {
         }
     }
 }
+
+impl From<B256> for BlockId {
+    fn from(value: B256) -> Self {
+        BlockId::Hash(value)
+    }
+}
+
+impl From<u32> for BlockId {
+    fn from(value: u32) -> Self {
+        BlockId::Slot(value)
+    }
+}
+
+impl From<BlockHeaderResponse> for BlockHeader {
+    fn from(response: BlockHeaderResponse) -> Self {
+        BlockHeader {
+            root: response.data.root,
+            parent_root: response.data.header.message.parent_root,
+            slot: response.data.header.message.slot,
+        }
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum BlockIdResolutionError {
+    #[error("Block with id '{0}' not found")]
+    BlockNotFound(BlockId),
+    #[error("Failed to resolve block id '{block_id}'")]
+    FailedBlockIdResolution {
+        block_id: BlockId,
+        #[source]
+        error: ClientError,
+    },
+}
+
+pub trait BlockIdResolution {
+    async fn resolve_to_slot(
+        &self,
+        beacon_client: &dyn CommonBeaconClient,
+    ) -> Result<u32, BlockIdResolutionError>;
+}
+
+impl BlockIdResolution for BlockId {
+    async fn resolve_to_slot(
+        &self,
+        beacon_client: &dyn CommonBeaconClient,
+    ) -> Result<u32, BlockIdResolutionError> {
+        match self {
+            BlockId::Slot(slot) => Ok(*slot),
+            _ => match beacon_client
+                .get_block_header(self.clone().into())
+                .await
+                .map_err(|err| BlockIdResolutionError::FailedBlockIdResolution {
+                    block_id: self.clone(),
+                    error: err,
+                })? {
+                Some(header) => Ok(header.slot),
+                None => Err(BlockIdResolutionError::BlockNotFound(self.clone())),
+            },
+        }
+    }
+}

src/clients/blobscan/mod.rs

@@ -1,23 +1,22 @@
 use std::fmt::Debug;
 
+use alloy::primitives::B256;
 use async_trait::async_trait;
 use backoff::ExponentialBackoff;
 use chrono::TimeDelta;
 use reqwest::{Client, Url};
 
 #[cfg(test)]
 use mockall::automock;
+use types::{BlobscanBlock, ReorgedBlocksRequestBody};
 
-use crate::{
-    clients::{blobscan::types::ReorgedSlotsResponse, common::ClientResult},
-    json_get, json_put,
-};
+use crate::{clients::common::ClientResult, json_get, json_put};
 
 use self::{
     jwt_manager::{Config as JWTManagerConfig, JWTManager},
     types::{
         Blob, Block, BlockchainSyncState, BlockchainSyncStateRequest, BlockchainSyncStateResponse,
-        IndexRequest, ReorgedSlotsRequest, Transaction,
+        IndexRequest, Transaction,
     },
 };
 
@@ -37,7 +36,12 @@ pub trait CommonBlobscanClient: Send + Sync + Debug {
         transactions: Vec<Transaction>,
         blobs: Vec<Blob>,
     ) -> ClientResult<()>;
-    async fn handle_reorged_slots(&self, slots: &[u32]) -> ClientResult<u32>;
+    async fn get_block(&self, slot: u32) -> ClientResult<Option<BlobscanBlock>>;
+    async fn handle_reorg(
+        &self,
+        rewinded_blocks: Vec<B256>,
+        forwarded_blocks: Vec<B256>,
+    ) -> ClientResult<()>;
     async fn update_sync_state(&self, sync_state: BlockchainSyncState) -> ClientResult<()>;
     async fn get_sync_state(&self) -> ClientResult<Option<BlockchainSyncState>>;
 }
@@ -93,15 +97,26 @@ impl CommonBlobscanClient for BlobscanClient {
         json_put!(&self.client, url, token, &req).map(|_: Option<()>| ())
     }
 
-    async fn handle_reorged_slots(&self, slots: &[u32]) -> ClientResult<u32> {
-        let url = self.base_url.join("indexer/reorged-slots")?;
+    async fn get_block(&self, slot: u32) -> ClientResult<Option<BlobscanBlock>> {
+        let url = self.base_url.join(&format!("slots/{}", slot))?;
+
+        json_get!(&self.client, url, BlobscanBlock, self.exp_backoff.clone())
+    }
+
+    async fn handle_reorg(
+        &self,
+        rewinded_blocks: Vec<B256>,
+        forwarded_blocks: Vec<B256>,
+    ) -> ClientResult<()> {
+        let url = self.base_url.join("indexer/reorged-blocks")?;
         let token = self.jwt_manager.get_token()?;
-        let req = ReorgedSlotsRequest {
-            reorged_slots: slots.to_owned(),
+
+        let req = ReorgedBlocksRequestBody {
+            forwarded_blocks,
+            rewinded_blocks,
         };
 
-        json_put!(&self.client, url, ReorgedSlotsResponse, token, &req)
-            .map(|res: Option<ReorgedSlotsResponse>| res.unwrap().total_updated_slots)
+        json_put!(&self.client, url, ReorgedBlocksRequestBody, token, &req).map(|_| ())
     }
 
     async fn update_sync_state(&self, sync_state: BlockchainSyncState) -> ClientResult<()> {

src/clients/blobscan/types.rs

@@ -8,6 +8,13 @@ use serde::{Deserialize, Serialize};
 
 use crate::{clients::beacon::types::Blob as BeaconBlob, utils::web3::calculate_versioned_hash};
 
+#[derive(Serialize, Deserialize, Debug)]
+pub struct BlobscanBlock {
+    pub hash: B256,
+    pub number: u32,
+    pub slot: u32,
+}
+
 #[derive(Serialize, Deserialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct Block {
@@ -86,16 +93,11 @@ pub struct IndexRequest {
     pub blobs: Vec<Blob>,
 }
 
-#[derive(Serialize, Debug)]
-#[serde(rename_all = "camelCase")]
-pub struct ReorgedSlotsRequest {
-    pub reorged_slots: Vec<u32>,
-}
-
-#[derive(Deserialize, Debug)]
+#[derive(Deserialize, Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
-pub struct ReorgedSlotsResponse {
-    pub total_updated_slots: u32,
+pub struct ReorgedBlocksRequestBody {
+    pub forwarded_blocks: Vec<B256>,
+    pub rewinded_blocks: Vec<B256>,
 }
 
 impl fmt::Debug for Blob {

src/indexer/error.rs

@@ -16,7 +16,7 @@ pub enum IndexerError {
     #[error(transparent)]
     SyncingTaskError(#[from] IndexingError),
     #[error("failed to retrieve blobscan's sync state")]
-    BlobscanSyncStateRetrievalError(#[source] ClientError),
+    BlobscanSyncStateRetrievalError(#[from] ClientError),
     #[error("failed to send syncing task message")]
     SyncingTaskMessageSendFailure(#[from] SendError<IndexerTaskMessage>),
 }

src/indexer/event_handlers/finalized_checkpoint.rs

@@ -3,8 +3,7 @@ use tracing::info;
 
 use crate::{
     clients::{
-        beacon::types::{BlockId, FinalizedCheckpointEventData},
-        blobscan::types::BlockchainSyncState,
+        beacon::types::FinalizedCheckpointEventData, blobscan::types::BlockchainSyncState,
         common::ClientError,
     },
     context::CommonContext,
@@ -46,7 +45,7 @@ where
         let last_finalized_block_number = match self
             .context
             .beacon_client()
-            .get_block(&BlockId::Hash(block_hash))
+            .get_block(block_hash.into())
             .await
             .map_err(|err| {
                 FinalizedCheckpointEventHandlerError::BlockRetrievalError(