diff --git a/src/clients/beacon/mod.rs b/src/clients/beacon/mod.rs index ea52ddc..28c0a3e 100644 --- a/src/clients/beacon/mod.rs +++ b/src/clients/beacon/mod.rs @@ -38,13 +38,7 @@ impl BeaconClient { } pub async fn get_block(&self, block_id: &BlockId) -> ClientResult> { - let block_id = match block_id { - BlockId::Hash(hash) => format!("0x{:x}", hash), - BlockId::Slot(slot) => slot.to_string(), - block_id => block_id.to_string(), - }; - - let path = format!("v2/beacon/blocks/{block_id}"); + let path = format!("v2/beacon/blocks/{}", { block_id.to_detailed_string() }); let url = self.base_url.join(path.as_str())?; json_get!(&self.client, url, BlockResponse, self.exp_backoff.clone()).map(|res| match res { @@ -54,7 +48,7 @@ impl BeaconClient { } pub async fn get_block_header(&self, block_id: &BlockId) -> ClientResult> { - let path = format!("v1/beacon/headers/{block_id}"); + let path = format!("v1/beacon/headers/{}", { block_id.to_detailed_string() }); let url = self.base_url.join(path.as_str())?; json_get!( @@ -70,7 +64,9 @@ impl BeaconClient { } pub async fn get_blobs(&self, block_id: &BlockId) -> ClientResult>> { - let path = format!("v1/beacon/blob_sidecars/{block_id}"); + let path = format!("v1/beacon/blob_sidecars/{}", { + block_id.to_detailed_string() + }); let url = self.base_url.join(path.as_str())?; json_get!(&self.client, url, BlobsResponse, self.exp_backoff.clone()).map(|res| match res { @@ -79,7 +75,7 @@ impl BeaconClient { }) } - pub fn subscribe_to_events(&self, topics: Vec) -> ClientResult { + pub fn subscribe_to_events(&self, topics: &[Topic]) -> ClientResult { let topics = topics .iter() .map(|topic| topic.into()) diff --git a/src/clients/beacon/types.rs b/src/clients/beacon/types.rs index d27dcfe..c165978 100644 --- a/src/clients/beacon/types.rs +++ b/src/clients/beacon/types.rs @@ -14,9 +14,11 @@ pub enum BlockId { } #[derive(Serialize, Debug)] +#[serde(rename_all = "snake_case")] pub enum Topic { Head, FinalizedCheckpoint, + ChainReorg, } #[derive(Deserialize, Debug)] @@ -85,7 +87,17 @@ pub struct BlockHeaderMessage { } #[derive(Deserialize, Debug)] -pub struct HeadBlockEventData { +pub struct ChainReorgEventData { + pub old_head_block: H256, + pub new_head_block: H256, + #[serde(deserialize_with = "deserialize_number")] + pub slot: u32, + #[serde(deserialize_with = "deserialize_number")] + pub depth: u32, +} + +#[derive(Deserialize, Debug)] +pub struct HeadEventData { #[serde(deserialize_with = "deserialize_number")] pub slot: u32, pub block: H256, @@ -105,6 +117,17 @@ where value.parse::().map_err(serde::de::Error::custom) } +impl BlockId { + pub fn to_detailed_string(&self) -> String { + match self { + BlockId::Head => String::from("head"), + BlockId::Finalized => String::from("finalized"), + BlockId::Slot(slot) => slot.to_string(), + BlockId::Hash(hash) => format!("0x{:x}", hash), + } + } +} + impl fmt::Display for BlockId { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { @@ -126,7 +149,16 @@ impl FromStr for BlockId { _ => match s.parse::() { Ok(num) => Ok(BlockId::Slot(num)), Err(_) => { - Err("Invalid block ID. Expected 'head', 'finalized' or a number.".to_string()) + if s.starts_with("0x") { + match H256::from_str(s) { + Ok(hash) => Ok(BlockId::Hash(hash)), + Err(_) => Err(format!("Invalid block ID hash: {s}")), + } + } else { + Err( + format!("Invalid block ID: {s}. Expected 'head', 'finalized', a hash or a number."), + ) + } } }, } @@ -136,14 +168,15 @@ impl FromStr for BlockId { impl From<&Topic> for String { fn from(value: &Topic) -> Self { match value { + Topic::ChainReorg => String::from("chain_reorg"), Topic::Head => String::from("head"), Topic::FinalizedCheckpoint => String::from("finalized_checkpoint"), } } } -impl From for BlockData { - fn from(event_data: HeadBlockEventData) -> Self { +impl From for BlockData { + fn from(event_data: HeadEventData) -> Self { Self { root: event_data.block, slot: event_data.slot, diff --git a/src/clients/blobscan/mod.rs b/src/clients/blobscan/mod.rs index 7b7e5de..d695360 100644 --- a/src/clients/blobscan/mod.rs +++ b/src/clients/blobscan/mod.rs @@ -2,13 +2,16 @@ use backoff::ExponentialBackoff; use chrono::TimeDelta; use reqwest::{Client, Url}; -use crate::{clients::common::ClientResult, json_get, json_put}; +use crate::{ + clients::{blobscan::types::ReorgedSlotsResponse, common::ClientResult}, + json_get, json_put, +}; use self::{ jwt_manager::{Config as JWTManagerConfig, JWTManager}, types::{ Blob, Block, BlockchainSyncState, BlockchainSyncStateRequest, BlockchainSyncStateResponse, - IndexRequest, ReorgedSlotRequest, Transaction, + IndexRequest, ReorgedSlotsRequest, Transaction, }, }; @@ -64,14 +67,15 @@ impl BlobscanClient { json_put!(&self.client, url, token, &req).map(|_: Option<()>| ()) } - pub async fn handle_reorged_slot(&self, slot: u32) -> ClientResult<()> { - let url = self.base_url.join("indexer/reorged-slot")?; + pub async fn handle_reorged_slots(&self, slots: &[u32]) -> ClientResult { + let url = self.base_url.join("indexer/reorged-slots")?; let token = self.jwt_manager.get_token()?; - let req = ReorgedSlotRequest { - new_head_slot: slot, + let req = ReorgedSlotsRequest { + reorged_slots: slots.to_owned(), }; - json_put!(&self.client, url, token, &req).map(|_: Option<()>| ()) + json_put!(&self.client, url, ReorgedSlotsResponse, token, &req) + .map(|res: Option| res.unwrap().total_updated_slots) } pub async fn update_sync_state(&self, sync_state: BlockchainSyncState) -> ClientResult<()> { diff --git a/src/clients/blobscan/types.rs b/src/clients/blobscan/types.rs index aa6c7cc..95ce669 100644 --- a/src/clients/blobscan/types.rs +++ b/src/clients/blobscan/types.rs @@ -87,8 +87,14 @@ pub struct IndexRequest { #[derive(Serialize, Debug)] #[serde(rename_all = "camelCase")] -pub struct ReorgedSlotRequest { - pub new_head_slot: u32, +pub struct ReorgedSlotsRequest { + pub reorged_slots: Vec, +} + +#[derive(Deserialize, Debug)] +#[serde(rename_all = "camelCase")] +pub struct ReorgedSlotsResponse { + pub total_updated_slots: u32, } impl fmt::Debug for Blob { diff --git a/src/clients/macros.rs b/src/clients/macros.rs index 609034b..662b9b2 100644 --- a/src/clients/macros.rs +++ b/src/clients/macros.rs @@ -95,7 +95,10 @@ macro_rules! json_get { /// Make a PUT request sending JSON. /// if JSON deser fails, emit a `WARN` level tracing event macro_rules! json_put { - ($client:expr, $url:expr, $auth_token:expr, $body:expr) => {{ + ($client:expr, $url:expr, $auth_token:expr, $body:expr) => { + json_put!($client, $url, (), $auth_token, $body) + }; + ($client:expr, $url:expr, $expected:ty, $auth_token:expr, $body:expr) => {{ let url = $url.clone(); let body = format!("{:?}", $body); @@ -123,7 +126,7 @@ macro_rules! json_put { }; let text = resp.text().await?; - let result: $crate::clients::common::ClientResponse<_> = text.parse()?; + let result: $crate::clients::common::ClientResponse<$expected> = text.parse()?; if result.is_err() { tracing::warn!( diff --git a/src/env.rs b/src/env.rs index adcb48c..4a5237f 100644 --- a/src/env.rs +++ b/src/env.rs @@ -1,8 +1,12 @@ use envy::Error::MissingValue; use serde::Deserialize; +use crate::network::Network; + #[derive(Deserialize, Debug)] pub struct Environment { + #[serde(default = "default_network")] + pub network_name: Network, #[serde(default = "default_blobscan_api_endpoint")] pub blobscan_api_endpoint: String, #[serde(default = "default_beacon_node_endpoint")] @@ -10,11 +14,14 @@ pub struct Environment { #[serde(default = "default_execution_node_endpoint")] pub execution_node_endpoint: String, pub secret_key: String, - #[serde(default = "default_dencun_fork_slot")] - pub dencun_fork_slot: u32, + pub dencun_fork_slot: Option, pub sentry_dsn: Option, } +fn default_network() -> Network { + Network::Devnet +} + fn default_blobscan_api_endpoint() -> String { "http://localhost:3001".to_string() } @@ -27,10 +34,6 @@ fn default_execution_node_endpoint() -> String { "http://localhost:8545".to_string() } -fn default_dencun_fork_slot() -> u32 { - 0 -} - impl Environment { pub fn from_env() -> Result { match envy::from_env::() { diff --git a/src/indexer/mod.rs b/src/indexer/mod.rs index 613c97c..6bc15a4 100644 --- a/src/indexer/mod.rs +++ b/src/indexer/mod.rs @@ -1,15 +1,18 @@ use std::thread; use anyhow::{anyhow, Context as AnyhowContext}; + use futures::StreamExt; use reqwest_eventsource::Event; use tokio::{sync::mpsc, task::JoinHandle}; -use tracing::{debug, error, info}; +use tracing::{debug, error, info, warn}; use crate::{ args::Args, clients::{ - beacon::types::{BlockId, FinalizedCheckpointEventData, HeadBlockEventData, Topic}, + beacon::types::{ + BlockId, ChainReorgEventData, FinalizedCheckpointEventData, HeadEventData, Topic, + }, blobscan::types::BlockchainSyncState, }, context::{Config as ContextConfig, Context}, @@ -42,6 +45,7 @@ impl Indexer { return Err(error.into()); } }; + let slots_checkpoint = args.slots_per_save; let num_threads = match args.num_threads { Some(num_threads) => num_threads, @@ -49,7 +53,9 @@ impl Indexer { .map_err(|err| anyhow!("Failed to get number of available threads: {:?}", err))? .get() as u32, }; - let dencun_fork_slot = env.dencun_fork_slot; + let dencun_fork_slot = env + .dencun_fork_slot + .unwrap_or(env.network_name.dencun_fork_slot()); Ok(Self { context, @@ -74,7 +80,10 @@ impl Indexer { None => match &sync_state { Some(state) => match state.last_lower_synced_slot { Some(slot) => BlockId::Slot(slot - 1), - None => BlockId::Head, + None => match state.last_upper_synced_slot { + Some(slot) => BlockId::Slot(slot - 1), + None => BlockId::Head, + }, }, None => BlockId::Head, }, @@ -84,7 +93,10 @@ impl Indexer { None => match &sync_state { Some(state) => match state.last_upper_synced_slot { Some(slot) => BlockId::Slot(slot + 1), - None => BlockId::Head, + None => match state.last_lower_synced_slot { + Some(slot) => BlockId::Slot(slot + 1), + None => BlockId::Head, + }, }, None => BlockId::Head, }, @@ -146,6 +158,8 @@ impl Indexer { tx: mpsc::Sender, start_block_id: BlockId, ) -> JoinHandle { + let task_name = "realtime_sync".to_string(); + let target = format!("indexer:{task_name}"); let task_context = self.context.clone(); let mut synchronizer = self._create_synchronizer(); @@ -153,21 +167,62 @@ impl Indexer { let result: Result<(), IndexerError> = async { let beacon_client = task_context.beacon_client(); let blobscan_client = task_context.blobscan_client(); + let topics = vec![ + Topic::ChainReorg, + Topic::Head, + Topic::FinalizedCheckpoint, + ]; let mut event_source = task_context .beacon_client() - .subscribe_to_events(vec![Topic::Head, Topic::FinalizedCheckpoint])?; + .subscribe_to_events(&topics)?; let mut is_initial_sync_to_head = true; while let Some(event) = event_source.next().await { match event { Ok(Event::Open) => { - debug!(target = "indexer", "Listening for head and finalized block events…") + let events = topics + .iter() + .map(|topic| topic.into()) + .collect::>() + .join(", "); + debug!(target, events, "Listening to beacon events…") } Ok(Event::Message(event)) => { - match event.event.as_str() { + let event_name = event.event.as_str(); + + match event_name { + "chain_reorg" => { + let reorg_block_data = + serde_json::from_str::(&event.data)?; + let slot = reorg_block_data.slot; + let old_head_block = reorg_block_data.old_head_block; + let target_depth = reorg_block_data.depth; + + let mut current_reorged_block = old_head_block; + let mut reorged_slots: Vec = vec![]; + + for current_depth in 1..=target_depth { + let reorged_block_head = match beacon_client.get_block_header(&BlockId::Hash(current_reorged_block)).await? { + Some(block) => block, + None => { + warn!(target, event=event_name, slot=slot, "Found {current_depth} out of {target_depth} reorged blocks only"); + break + } + }; + + reorged_slots.push(reorged_block_head.header.message.slot); + current_reorged_block = reorged_block_head.header.message.parent_root; + } + + let total_updated_slots = blobscan_client.handle_reorged_slots(&reorged_slots).await?; + + info!(target, event=event_name, slot=slot, "Reorganization of depth {target_depth} detected. Found the following reorged slots: {:#?}. Total slots marked as reorged: {total_updated_slots}", reorged_slots); + + panic!("Chain reorg detected"); + }, "head" => { let head_block_data = - serde_json::from_str::(&event.data)?; + serde_json::from_str::(&event.data)?; let head_block_id = &BlockId::Slot(head_block_data.slot); let initial_block_id = if is_initial_sync_to_head { @@ -215,7 +270,7 @@ impl Indexer { }) .await?; - info!(target = "indexer", "Finalized block {full_block_hash} detected and stored"); + info!(target, event=event_name, execution_block=last_finalized_block_number, "New finalized block detected"); }, unexpected_event_id => { return Err(IndexerError::UnexpectedEvent { event: unexpected_event_id.to_string() }) @@ -237,7 +292,7 @@ impl Indexer { if let Err(error) = result { // TODO: Find a better way to handle this error tx.send(Err(IndexingTaskError::FailedIndexingTask { - task_name: "realtime_head_block_sync".to_string(), + task_name, error, })) .await diff --git a/src/main.rs b/src/main.rs index ac83c90..cf3d8cd 100644 --- a/src/main.rs +++ b/src/main.rs @@ -11,6 +11,7 @@ mod clients; mod context; mod env; mod indexer; +mod network; mod slots_processor; mod synchronizer; mod utils; @@ -35,16 +36,21 @@ pub fn print_banner(args: &Args, env: &Environment) { println!("Blobscan indexer (EIP-4844 blob indexer) - blobscan.com"); println!("======================================================="); - if let Some(from_slot) = args.from_slot.clone() { - println!("Start slot: {}", from_slot); + println!("Network: {:?}", env.network_name); + if let Some(dencun_fork_slot) = env.dencun_fork_slot { + println!("Dencun fork slot: {dencun_fork_slot}"); } else { - println!("Start slot: 0"); + println!("Dencun fork slot: {}", env.network_name.dencun_fork_slot()); + } + + if let Some(from_slot) = args.from_slot.clone() { + println!("Custom start slot: {}", from_slot.to_detailed_string()); } if let Some(num_threads) = args.num_threads { println!("Number of threads: {}", num_threads); } else { - println!("Number of threads: 1"); + println!("Number of threads: auto"); } if let Some(slots_per_save) = args.slots_per_save { @@ -53,7 +59,6 @@ pub fn print_banner(args: &Args, env: &Environment) { println!("Slots checkpoint size: 1000"); } - println!("Dencun fork slot: {}", env.dencun_fork_slot); println!("Blobscan API endpoint: {}", env.blobscan_api_endpoint); println!( "CL endpoint: {:?}", diff --git a/src/network.rs b/src/network.rs new file mode 100644 index 0000000..b1046eb --- /dev/null +++ b/src/network.rs @@ -0,0 +1,23 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Deserialize, Serialize)] +#[serde(rename_all = "lowercase")] +pub enum Network { + Mainnet, + Goerli, + Sepolia, + Holesky, + Devnet, +} + +impl Network { + pub fn dencun_fork_slot(&self) -> u32 { + match self { + Network::Mainnet => 8626176, + Network::Goerli => 7413760, + Network::Sepolia => 4243456, + Network::Holesky => 950272, + Network::Devnet => 0, + } + } +} diff --git a/src/slots_processor/mod.rs b/src/slots_processor/mod.rs index f6cfcd3..c0b436e 100644 --- a/src/slots_processor/mod.rs +++ b/src/slots_processor/mod.rs @@ -19,7 +19,6 @@ mod helpers; pub struct SlotsProcessor { context: Context, - last_block: Option, } #[derive(Debug, Clone)] @@ -38,11 +37,8 @@ impl From for BlockData { } impl SlotsProcessor { - pub fn new(context: Context, last_block: Option) -> SlotsProcessor { - Self { - context, - last_block, - } + pub fn new(context: Context) -> SlotsProcessor { + Self { context } } pub async fn process_slots( @@ -54,7 +50,7 @@ impl SlotsProcessor { if is_reverse_processing { for current_slot in (final_slot..=initial_slot).rev() { - let result = self.process_slot(current_slot, Some(false)).await; + let result = self.process_slot(current_slot).await; if let Err(error) = result { return Err(SlotsProcessorError::FailedSlotsProcessing { @@ -67,7 +63,7 @@ impl SlotsProcessor { } } else { for current_slot in initial_slot..=final_slot { - let result = self.process_slot(current_slot, Some(true)).await; + let result = self.process_slot(current_slot).await; if let Err(error) = result { return Err(SlotsProcessorError::FailedSlotsProcessing { @@ -83,17 +79,7 @@ impl SlotsProcessor { Ok(()) } - pub async fn process_slot( - &mut self, - slot: u32, - enable_reorg_detection: Option, - ) -> Result<(), SlotProcessingError> { - if let Some(enable_reorg_detection) = enable_reorg_detection { - if enable_reorg_detection { - self._detect_and_handle_reorg(slot).await?; - } - } - + pub async fn process_slot(&mut self, slot: u32) -> Result<(), SlotProcessingError> { let beacon_client = self.context.beacon_client(); let blobscan_client = self.context.blobscan_client(); let provider = self.context.provider(); @@ -227,38 +213,4 @@ impl SlotsProcessor { Ok(()) } - - pub fn get_last_block(&self) -> Option { - self.last_block.clone() - } - - async fn _detect_and_handle_reorg(&mut self, slot: u32) -> Result<(), SlotProcessingError> { - let beacon_client = self.context.beacon_client(); - let blobscan_client = self.context.blobscan_client(); - - let beacon_block_header = match beacon_client.get_block_header(&BlockId::Slot(slot)).await? - { - Some(block_header) => block_header, - None => { - debug!( - target = "slots_processor", - slot, "Skipping as there is no beacon block header" - ); - - return Ok(()); - } - }; - - if let Some(block) = &self.last_block { - if beacon_block_header.header.message.parent_root != block.root { - info!(target = "slots_processor", slot, "Block reorg detected"); - - blobscan_client.handle_reorged_slot(slot).await?; - } - } - - self.last_block = Some(beacon_block_header.into()); - - Ok(()) - } } diff --git a/src/synchronizer/mod.rs b/src/synchronizer/mod.rs index 1228802..2e71876 100644 --- a/src/synchronizer/mod.rs +++ b/src/synchronizer/mod.rs @@ -6,7 +6,7 @@ use tracing::{debug, debug_span, info, Instrument}; use crate::{ clients::{beacon::types::BlockId, blobscan::types::BlockchainSyncState, common::ClientError}, context::Context, - slots_processor::{error::SlotsProcessorError, BlockData, SlotsProcessor}, + slots_processor::{error::SlotsProcessorError, SlotsProcessor}, }; use self::error::{SlotsChunksErrors, SynchronizerError}; @@ -25,7 +25,6 @@ pub struct Synchronizer { num_threads: u32, min_slots_per_thread: u32, slots_checkpoint: u32, - last_synced_block: Option, } impl Default for SynchronizerBuilder { @@ -60,7 +59,6 @@ impl SynchronizerBuilder { num_threads: self.num_threads, min_slots_per_thread: self.min_slots_per_thread, slots_checkpoint: self.slots_checkpoint, - last_synced_block: None, } } } @@ -98,11 +96,10 @@ impl Synchronizer { let num_threads = std::cmp::max(1, unprocessed_slots / slots_per_thread); let remaining_slots = unprocessed_slots % num_threads; - let mut handles: Vec, SlotsProcessorError>>> = vec![]; + let mut handles: Vec>> = vec![]; for i in 0..num_threads { - let mut slots_processor = - SlotsProcessor::new(self.context.clone(), self.last_synced_block.clone()); + let mut slots_processor = SlotsProcessor::new(self.context.clone()); let thread_total_slots = slots_per_thread + if i == num_threads - 1 { remaining_slots @@ -132,7 +129,7 @@ impl Synchronizer { .process_slots(thread_initial_slot, thread_final_slot) .await?; - Ok(slots_processor.get_last_block()) + Ok(()) } .instrument(synchronizer_thread_span), ); @@ -143,14 +140,11 @@ impl Synchronizer { let handle_outputs = join_all(handles).await; let mut errors = vec![]; - let mut last_synced_block: Option = None; for handle in handle_outputs { match handle { Ok(thread_result) => match thread_result { - Ok(thread_last_block) => { - last_synced_block = thread_last_block; - } + Ok(()) => {} Err(error) => errors.push(error), }, Err(error) => { @@ -162,8 +156,6 @@ impl Synchronizer { } if errors.is_empty() { - self.last_synced_block = last_synced_block; - Ok(()) } else { Err(SynchronizerError::FailedParallelSlotsProcessing {