fix(hubble): arbitrum consensus height tracking (#3337)

Author: qlp

hubble/src/arb.rs

@@ -11,7 +11,7 @@ use alloy::{
 };
 use backon::{ConstantBuilder, ExponentialBuilder, Retryable};
 use color_eyre::eyre::{eyre, ContextCompat, Result, WrapErr};
-use tracing::{debug, info};
+use tracing::{debug, info, trace};
 use unionlabs::{bounded::BoundedU32, hash::H160, uint::U256};
 
 use crate::{
@@ -86,6 +86,8 @@ impl Config {
         .retry(&ExponentialBuilder::default())
         .await?;
 
+        info!("fetched db chain_id for chain {l2_chain_id} => {chain_id:?}");
+
         let querier = Arb {
             l1_client: ProviderBuilder::new().on_http(self.l1_url),
             l2_client,
@@ -102,8 +104,13 @@ impl Config {
 impl Arb {
     // NOTE: Copied from chain_utils
     async fn execution_height_of_beacon_slot(&self, slot: u64) -> Result<u64> {
+        trace!("find execution height of beacon slot {slot}");
+
         // read the next_node_num at l1.execution_height(beacon_slot), then from there filter for `NodeCreated`
         let next_node_num = self.next_node_num_at_beacon_slot(slot).await?;
+
+        trace!("find execution height of beacon slot {slot}. next node num: {next_node_num}");
+
         let [event] = self
             .l1_client
             .get_logs(
@@ -123,24 +130,38 @@ impl Arb {
             .await
             .wrap_err("error fetching `NodeCreated` log from l1")?
             .try_into()
-            .map_err(|e| eyre!("too many logs found? there should only be one `NodeCreated event`, but found: {e:?}"))?;
-        debug!("next node num: {next_node_num}: {event:?}");
+            .map_err(|e| eyre!("too many logs or no found? there should only be one `NodeCreated event`, but found: {e:?}"))?;
+
+        trace!("find execution height of beacon slot {slot}. event: {event:?}");
+
         let event = NodeCreated::decode_log(&event.inner, true).unwrap();
+
+        trace!("find execution height of beacon slot {slot}. event(decoded): {event:?}");
         let block_id = BlockId::Hash(RpcBlockHash {
             block_hash: FixedBytes::from_slice(event.assertion.0 .0 .0[0].as_ref()),
             require_canonical: None,
         });
+
+        trace!("find execution height of beacon slot {slot}. block-id: {block_id}");
+
         let block = self
             .l2_client
-            .get_block(block_id, BlockTransactionsKind::Full)
+            .get_block(block_id, BlockTransactionsKind::Hashes)
             .await
             .wrap_err("error fetching l2 block")?
             .expect("block should exist if it is finalized on the l1");
 
+        trace!(
+            "find execution height of beacon slot {slot}. block-number: {}",
+            block.header.number
+        );
+
         Ok(block.header.number)
     }
 
     pub async fn next_node_num_at_beacon_slot(&self, slot: u64) -> Result<u64> {
+        trace!("find next node num at beacon slot {slot}");
+
         let l1_height = self
             .beacon
             .get_height_at_skip_missing(slot.try_into().expect("negative slot?"))
@@ -151,12 +172,15 @@ impl Arb {
             .execution_payload
             .block_number;
 
+        trace!("find next node num at beacon slot {slot}: l1-height: {l1_height}");
+
         let slot_offset_bytes = self
             .rollup_finalization_config
             .l1_next_node_num_slot_offset_bytes
             .inner()
             .try_into()
             .unwrap();
+
         let raw_slot = self
             .l1_client
             .get_storage_at(
@@ -171,27 +195,41 @@ impl Arb {
             )
             .await?;
 
+        trace!("find next node num at beacon slot {slot}: l1-height: {l1_height}: raw_slow: {raw_slot}");
+
         let raw_slot: B256 = raw_slot.into();
         let latest_confirmed = u64::from_be_bytes(
             raw_slot.0[slot_offset_bytes..slot_offset_bytes + 8]
                 .try_into()
                 .expect("size is correct; qed;"),
         );
 
+        trace!("find next node num at beacon slot {slot}: l1-height: {l1_height}: latest_confirmed: {latest_confirmed}");
+
         debug!("l1_height {l1_height} is next node num {latest_confirmed}",);
         Ok(latest_confirmed)
     }
 }
 
 impl Querier for Arb {
     async fn get_execution_height(&self, slot: i64) -> Result<(i64, i64)> {
+        trace!("get execution height of beacon slot {slot}");
+
         let height = (|| self.execution_height_of_beacon_slot(slot.try_into().unwrap()))
             .retry(
                 &ConstantBuilder::default()
                     .with_delay(Duration::from_millis(500))
                     .with_max_times(60),
             )
+            .notify(|err, duration| {
+                trace!(
+                    "get execution height of beacon slot {slot} => error: {err:?}. retry after {}s",
+                    duration.as_secs()
+                );
+            })
             .await?;
+
+        trace!("get execution height of beacon slot {slot}, found: {height}");
         Ok((slot, height.try_into().unwrap()))
     }
 }

hubble/src/indexer/eth/block_handle.rs

@@ -1,4 +1,4 @@
-use alloy::rpc::types::Block;
+use alloy::network::AnyRpcBlock;
 use axum::async_trait;
 use color_eyre::eyre::Report;
 use futures::{stream::FuturesOrdered, Stream};
@@ -21,7 +21,7 @@ use crate::{
 
 #[derive(Clone)]
 pub enum BlockDetails {
-    Lazy(Block),
+    Lazy(AnyRpcBlock),
     Eager(Option<BlockInsert>),
 }
 
@@ -30,7 +30,7 @@ pub enum BlockDetails {
 pub struct BlockInsert {
     pub chain_id: ChainId,
     pub hash: String,
-    pub header: Block,
+    pub header: AnyRpcBlock,
     pub height: i32,
     pub time: OffsetDateTime,
     pub transactions: Vec<TransactionInsert>,

hubble/src/indexer/eth/fetcher_client.rs

@@ -2,8 +2,9 @@ use std::{collections::HashMap, fmt::Display};
 
 use alloy::{
     eips::BlockId,
+    network::AnyRpcBlock,
     primitives::{Address, BloomInput, FixedBytes},
-    rpc::types::{Block, BlockTransactionsKind, Filter, Log},
+    rpc::types::{BlockTransactionsKind, Filter, Log},
 };
 use axum::async_trait;
 use color_eyre::eyre::Report;
@@ -45,7 +46,7 @@ trait BlockReferenceProvider {
     fn block_reference(&self) -> Result<BlockReference, Report>;
 }
 
-impl BlockReferenceProvider for Block {
+impl BlockReferenceProvider for AnyRpcBlock {
     fn block_reference(&self) -> Result<BlockReference, Report> {
         Ok(BlockReference {
             height: self.header.number,
@@ -110,6 +111,8 @@ impl EthFetcherClient {
             )
             .await;
 
+        info!("block: {block:?}");
+
         match block {
             Ok(rpc_result) => match rpc_result {
                 Some(result) => {
@@ -147,7 +150,7 @@ impl EthFetcherClient {
 
     pub async fn fetch_details(
         &self,
-        block: &Block,
+        block: &AnyRpcBlock,
         provider_id: RpcProviderId,
     ) -> Result<Option<BlockInsert>, IndexerError> {
         let block_reference = block.block_reference()?;

hubble/src/indexer/eth/postgres.rs

@@ -1,4 +1,4 @@
-use alloy::rpc::types::Block;
+use alloy::network::AnyRpcBlock;
 use itertools::Itertools;
 use serde::{Deserialize, Serialize};
 use sqlx::{PgPool, Postgres, Transaction};
@@ -26,7 +26,7 @@ pub struct PgLog {
 #[derive(Serialize, Deserialize)]
 pub struct PgLogData {
     pub transactions: Vec<TransactionInsert>,
-    pub header: Block,
+    pub header: AnyRpcBlock,
 }
 
 impl From<BlockInsert> for PgLog {

hubble/src/indexer/eth/provider.rs

@@ -1,9 +1,9 @@
 use alloy::{
     eips::BlockId,
-    network::{Ethereum, Network},
+    network::{AnyNetwork, AnyRpcBlock, Network},
     primitives::TxHash,
     providers::{Provider as AlloyProvider, ProviderBuilder, RootProvider},
-    rpc::types::{Block, BlockTransactionsKind, Filter, Log},
+    rpc::types::{BlockTransactionsKind, Filter, Log},
     transports::{
         http::{Client, Http},
         RpcError, TransportErrorKind,
@@ -15,7 +15,7 @@ use crate::race_client::{RaceClient, RaceClientId, RaceClientResponse};
 
 #[derive(Clone, Debug)]
 pub struct Provider {
-    pub rpc_client: RaceClient<RootProvider<Http<Client>>>,
+    pub rpc_client: RaceClient<RootProvider<Http<Client>, AnyNetwork>>,
 }
 
 #[derive(Clone, Debug, Copy)]
@@ -56,7 +56,7 @@ impl Provider {
             rpc_client: RaceClient::new(
                 rpc_urls
                     .into_iter()
-                    .map(|url| ProviderBuilder::new().on_http(url))
+                    .map(|url| ProviderBuilder::new().network::<AnyNetwork>().on_http(url))
                     .collect(),
             ),
         }
@@ -77,7 +77,7 @@ impl Provider {
         id: BlockId,
         kind: BlockTransactionsKind,
         provider_id: Option<RpcProviderId>,
-    ) -> Result<Option<RpcResult<Block>>, RpcError<TransportErrorKind>> {
+    ) -> Result<Option<RpcResult<AnyRpcBlock>>, RpcError<TransportErrorKind>> {
         self.rpc_client
             .race_some(provider_id.map(Into::into), |c| c.get_block(id, kind))
             .await
@@ -100,7 +100,7 @@ impl Provider {
         tx_hash: TxHash,
         provider_id: Option<RpcProviderId>,
     ) -> Result<
-        Option<RpcResult<<Ethereum as Network>::TransactionResponse>>,
+        Option<RpcResult<<AnyNetwork as Network>::TransactionResponse>>,
         RpcError<TransportErrorKind>,
     > {
         self.rpc_client