Handle reorgs

Cargo.toml

@@ -13,6 +13,7 @@ ethers = "1.0.2"
 futures = "0.3.25"
 hex = "0.4.3"
 reqwest = { version = "0.11.13", features = ["json"] }
+reqwest-eventsource = "0.5.0"
 url = { version = "2.3.1", features = ["serde"] }
 serde = { version = "1.0.150", features = ["derive"] }
 tokio = { version = "1.23.0", features = ["full"] }

src/chain_reorg_handler/mod.rs

@@ -0,0 +1,103 @@
+use anyhow::Result;
+use futures::{future::join_all, StreamExt};
+use reqwest_eventsource::Event;
+use tokio::task::JoinHandle;
+use tracing::{debug, error, Instrument};
+
+use crate::{
+    clients::beacon::types::{ChainReorgResponse, Topic},
+    context::Context,
+    slots_processor::slot_processor::SlotProcessor,
+};
+
+pub struct ChainReorgHandler {
+    context: Context,
+}
+
+impl ChainReorgHandler {
+    pub fn new(context: Context) -> ChainReorgHandler {
+        Self { context }
+    }
+
+    pub fn run(&self) -> Result<JoinHandle<Result<(), anyhow::Error>>> {
+        let thread_context = self.context.clone();
+
+        let handle = tokio::spawn(async move {
+            let blobscan_client = thread_context.blobscan_client();
+            let beacon_client = thread_context.beacon_client();
+            let mut event_source = match beacon_client.subscribe_to_events(vec![Topic::ChainReorg])
+            {
+                Ok(es) => es,
+                Err(error) => {
+                    error!(
+                        target = "chain_reorg_handler",
+                        ?error,
+                        "Failed to subscribe to chain reorg events"
+                    );
+
+                    return Err(error.into());
+                }
+            };
+
+            while let Some(event) = event_source.next().await {
+                match event {
+                    Ok(Event::Open) => debug!(
+                        target = "chain_reorg_handler",
+                        "Listening to chain reorg events…"
+                    ),
+                    Ok(Event::Message(message)) => {
+                        let chain_reorg_data =
+                            serde_json::from_str::<ChainReorgResponse>(&message.data)?;
+
+                        let new_head_slot = chain_reorg_data.slot.parse::<u32>()?;
+                        let reorg_depth = chain_reorg_data.depth.parse::<u32>()?;
+
+                        blobscan_client
+                            .handle_reorg(new_head_slot, reorg_depth)
+                            .await?;
+
+                        // reprocess reorganized slots
+                        let slot_processor = SlotProcessor::new(thread_context.clone());
+
+                        let mut futures = Vec::new();
+
+                        for slot in new_head_slot..new_head_slot + reorg_depth {
+                            let slot_span = tracing::trace_span!("slot_processor");
+
+                            futures.push(slot_processor.process_slot(slot).instrument(slot_span));
+                        }
+
+                        let results = join_all(futures).await;
+
+                        for result in results {
+                            if let Err(error) = result {
+                                error!(
+                                    target = "chain_reorg_handler",
+                                    ?error,
+                                    "Failed to process reorg slot"
+                                );
+
+                                return Err(error.into());
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        error!(
+                            target = "chain_reorg_handler",
+                            ?e,
+                            "Failed to receive chain reorg event"
+                        );
+
+                        event_source.close();
+
+                        return Err(e.into());
+                    }
+                }
+            }
+
+            Ok(())
+        });
+
+        Ok(handle)
+    }
+}

src/clients/beacon/mod.rs

@@ -1,10 +1,11 @@
 use anyhow::Context as AnyhowContext;
 use reqwest::{Client, Url};
+use reqwest_eventsource::EventSource;
 use std::time::Duration;
 
 use crate::{clients::common::ClientResult, json_get};
 
-use self::types::{Blob, BlobsResponse, BlockMessage as Block, BlockResponse};
+use self::types::{Blob, BlobsResponse, BlockMessage as Block, BlockResponse, Topic};
 
 pub mod types;
 
@@ -50,4 +51,18 @@ impl BeaconClient {
             None => None,
         })
     }
+
+    pub fn subscribe_to_events(&self, topics: Vec<Topic>) -> ClientResult<EventSource> {
+        let topics = topics
+            .iter()
+            .map(|topic| topic.to_string())
+            .collect::<Vec<String>>()
+            .join("&");
+        let path = format!("v1/events?topics={topics}");
+        let url = self.base_url.join(&path)?;
+
+        let event_source = EventSource::get(url);
+
+        Ok(event_source)
+    }
 }

src/clients/beacon/types.rs

@@ -1,6 +1,18 @@
 use ethers::types::{Bytes, H256};
 use serde::Deserialize;
 
+pub enum Topic {
+    ChainReorg,
+}
+
+impl Topic {
+    pub fn to_string(&self) -> String {
+        match self {
+            Topic::ChainReorg => String::from("chain_reorg"),
+        }
+    }
+}
+
 #[derive(Deserialize, Debug)]
 pub struct ExecutionPayload {
     pub block_hash: H256,
@@ -38,3 +50,11 @@
 pub struct BlobsResponse {
     pub data: Vec<Blob>,
 }
+
+#[derive(Deserialize, Debug)]
+pub struct ChainReorgResponse {
+    pub slot: String,
+    pub depth: String,
+    pub old_head_block: String,
+    pub new_head_block: String,
+}

src/clients/blobscan/mod.rs

@@ -6,7 +6,9 @@ use crate::{clients::common::ClientResult, json_get, json_put};
 
 use self::{
     jwt_manager::{Config as JWTManagerConfig, JWTManager},
-    types::{Blob, Block, IndexRequest, SlotRequest, SlotResponse, Transaction},
+    types::{
+        Blob, Block, IndexRequest, ReorgedSlotRequest, SlotRequest, SlotResponse, Transaction,
+    },
 };
 
 mod jwt_manager;
@@ -58,10 +60,18 @@ impl BlobscanClient {
         json_put!(&self.client, url, token, &req).map(|_: Option<()>| ())
     }
 
+    pub async fn handle_reorg(&self, slot: u32, depth: u32) -> ClientResult<()> {
+        let url = self.base_url.join("reorged-slot")?;
+        let token = self.jwt_manager.get_token()?;
+        let req = ReorgedSlotRequest { slot, depth };
+
+        json_put!(&self.client, url, token, &req).map(|_: Option<()>| ())
+    }
+
     pub async fn update_slot(&self, slot: u32) -> ClientResult<()> {
+        let url = self.base_url.join("slot")?;
         let token = self.jwt_manager.get_token()?;
         let req = SlotRequest { slot };
-        let url = self.base_url.join("slot")?;
 
         json_put!(&self.client, url, token, &req).map(|_: Option<()>| ())
     }

src/clients/blobscan/types.rs

@@ -67,6 +67,12 @@ pub struct IndexRequest {
     pub blobs: Vec<Blob>,
 }
 
+#[derive(Serialize, Debug)]
+pub struct ReorgedSlotRequest {
+    pub slot: u32,
+    pub depth: u32,
+}
+
 impl fmt::Debug for Blob {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(

src/indexer.rs

@@ -11,6 +11,7 @@ use tracing::{debug, error, info, warn, Instrument};
 
 use crate::{
     args::Args,
+    chain_reorg_handler::ChainReorgHandler,
     context::{Config as ContextConfig, Context},
     env::Environment,
     slots_processor::{Config as SlotsProcessorConfig, SlotsProcessor},
@@ -58,6 +59,10 @@ pub async fn run(env: Environment) -> Result<()> {
         }
     };
 
+    let chain_reorg_handler = ChainReorgHandler::new(context.clone());
+
+    let _handler = chain_reorg_handler.run();
+
     let beacon_client = context.beacon_client();
     let blobscan_client = context.blobscan_client();
 

src/main.rs

@@ -3,6 +3,7 @@ use env::Environment;
 use utils::telemetry::{get_subscriber, init_subscriber};
 
 mod args;
+mod chain_reorg_handler;
 mod clients;
 mod context;
 mod env;

src/slots_processor/mod.rs

@@ -10,7 +10,7 @@ use self::{
 use crate::context::Context;
 
 mod error;
-mod slot_processor;
+pub mod slot_processor;
 
 pub struct SlotsProcessor {
     context: Context,