chain/ethereum: migrate block methods to alloy part 1

Author: incrypto32

chain/ethereum/src/adapter.rs

@@ -7,6 +7,8 @@ use graph::data_source::common::ContractCall;
 use graph::firehose::CallToFilter;
 use graph::firehose::CombinedFilter;
 use graph::firehose::LogFilter;
+use graph::prelude::alloy::primitives::B256;
+use graph::prelude::alloy::rpc::types::Block as AlloyBlock;
 use graph::prelude::alloy::transports::{RpcError, TransportErrorKind};
 use graph::prelude::web3::types::H160;
 use itertools::Itertools;
@@ -1244,12 +1246,6 @@ pub trait EthereumAdapter: Send + Sync + 'static {
     /// Get the latest block, with only the header and transaction hashes.
     async fn latest_block_ptr(&self, logger: &Logger) -> Result<BlockPtr, bc::IngestorError>;
 
-    async fn load_block(
-        &self,
-        logger: &Logger,
-        block_hash: H256,
-    ) -> Result<LightEthereumBlock, Error>;
-
     /// Load Ethereum blocks in bulk, returning results as they come back as a Stream.
     /// May use the `chain_store` as a cache.
     async fn load_blocks(
@@ -1263,20 +1259,20 @@ pub trait EthereumAdapter: Send + Sync + 'static {
     async fn block_by_hash(
         &self,
         logger: &Logger,
-        block_hash: H256,
-    ) -> Result<Option<LightEthereumBlock>, Error>;
+        block_hash: B256,
+    ) -> Result<Option<AlloyBlock>, Error>;
 
     async fn block_by_number(
         &self,
         logger: &Logger,
         block_number: BlockNumber,
-    ) -> Result<Option<LightEthereumBlock>, Error>;
+    ) -> Result<Option<AlloyBlock>, Error>;
 
     /// Load full information for the specified `block` (in particular, transaction receipts).
     async fn load_full_block(
         &self,
         logger: &Logger,
-        block: LightEthereumBlock,
+        block: AlloyBlock,
     ) -> Result<EthereumBlock, bc::IngestorError>;
 
     /// Finds the hash and number of the lowest non-null block with height greater than or equal to

chain/ethereum/src/ethereum_adapter.rs

@@ -32,17 +32,18 @@ use graph::prelude::{
         },
         rpc::types::{
             trace::{filter::TraceFilter as AlloyTraceFilter, parity::LocalizedTransactionTrace},
-            TransactionInput, TransactionRequest,
+            Block as AlloyBlock, TransactionInput, TransactionRequest,
         },
         transports::{RpcError, TransportErrorKind},
     },
-    alloy_log_to_web3_log, alloy_transaction_receipt_to_web3_transaction_receipt, b256_to_h256,
+    alloy_log_to_web3_log, alloy_transaction_receipt_to_web3_transaction_receipt,
     h160_to_alloy_address, h256_to_b256,
     tokio::try_join,
 };
 use graph::slog::o;
 use graph::tokio::sync::RwLock;
 use graph::tokio::time::timeout;
+use graph::util::conversions::alloy_block_to_web3_block;
 use graph::{
     blockchain::{block_stream::BlockWithTriggers, BlockPtr, IngestorError},
     prelude::{
@@ -1258,27 +1259,12 @@ impl EthereumAdapterTrait for EthereumAdapter {
             .await
     }
 
-    async fn load_block(
-        &self,
-        logger: &Logger,
-        block_hash: H256,
-    ) -> Result<LightEthereumBlock, Error> {
-        self.block_by_hash(logger, block_hash)
-            .await?
-            .ok_or_else(move || {
-                anyhow!(
-                    "Ethereum node could not find block with hash {}",
-                    block_hash
-                )
-            })
-    }
-
     async fn block_by_hash(
         &self,
         logger: &Logger,
-        block_hash: H256,
-    ) -> Result<Option<LightEthereumBlock>, Error> {
-        let web3 = self.web3.clone();
+        block_hash: B256,
+    ) -> Result<Option<AlloyBlock>, Error> {
+        let alloy = self.alloy.clone();
         let logger = logger.clone();
         let retry_log_message = format!(
             "eth_getBlockByHash RPC call for block hash {:?}",
@@ -1290,10 +1276,11 @@ impl EthereumAdapterTrait for EthereumAdapter {
             .limit(ENV_VARS.request_retries)
             .timeout_secs(ENV_VARS.json_rpc_timeout.as_secs())
             .run(move || {
-                let web3 = web3.cheap_clone();
+                let alloy = alloy.cheap_clone();
                 async move {
-                    web3.eth()
-                        .block_with_txs(BlockId::Hash(block_hash))
+                    alloy
+                        .get_block_by_hash(block_hash)
+                        .full()
                         .await
                         .map_err(Error::from)
                 }
@@ -1310,8 +1297,8 @@ impl EthereumAdapterTrait for EthereumAdapter {
         &self,
         logger: &Logger,
         block_number: BlockNumber,
-    ) -> Result<Option<LightEthereumBlock>, Error> {
-        let web3 = self.web3.clone();
+    ) -> Result<Option<AlloyBlock>, Error> {
+        let alloy = self.alloy.clone();
         let logger = logger.clone();
         let retry_log_message = format!(
             "eth_getBlockByNumber RPC call for block number {}",
@@ -1322,10 +1309,13 @@ impl EthereumAdapterTrait for EthereumAdapter {
             .no_limit()
             .timeout_secs(ENV_VARS.json_rpc_timeout.as_secs())
             .run(move || {
-                let web3 = web3.cheap_clone();
+                let alloy = alloy.clone();
                 async move {
-                    web3.eth()
-                        .block_with_txs(BlockId::Number(block_number.into()))
+                    alloy
+                        .get_block_by_number(alloy::rpc::types::BlockNumberOrTag::Number(
+                            block_number as u64,
+                        ))
+                        .full()
                         .await
                         .map_err(Error::from)
                 }
@@ -1344,50 +1334,42 @@ impl EthereumAdapterTrait for EthereumAdapter {
     async fn load_full_block(
         &self,
         logger: &Logger,
-        block: LightEthereumBlock,
+        block: AlloyBlock,
     ) -> Result<EthereumBlock, IngestorError> {
         let alloy = self.alloy.clone();
         let logger = logger.clone();
-        let block_hash = block.hash.expect("block is missing block hash");
+        let block_hash = block.header.hash;
 
         // The early return is necessary for correctness, otherwise we'll
         // request an empty batch which is not valid in JSON-RPC.
         if block.transactions.is_empty() {
             trace!(logger, "Block {} contains no transactions", block_hash);
             return Ok(EthereumBlock {
-                block: Arc::new(block),
+                block: Arc::new(alloy_block_to_web3_block(block)),
                 transaction_receipts: Vec::new(),
             });
         }
-        let hashes: Vec<_> = block.transactions.iter().map(|txn| txn.hash).collect();
+        let hashes: Vec<_> = block.transactions.hashes().collect();
 
         let supports_block_receipts = self
             .check_block_receipt_support_and_update_cache(
                 alloy.clone(),
-                h256_to_b256(block_hash),
+                block_hash,
                 self.supports_eip_1898,
                 self.call_only,
                 logger.clone(),
             )
             .await;
 
-        let hashes_b256 = hashes.iter().map(|hash| h256_to_b256(*hash)).collect();
-        let block_hash_b256 = h256_to_b256(block_hash);
-        fetch_receipts_with_retry(
-            alloy,
-            hashes_b256,
-            block_hash_b256,
-            logger,
-            supports_block_receipts,
-        )
-        .await
-        .map(|transaction_receipts| EthereumBlock {
-            block: Arc::new(block),
-            transaction_receipts: transaction_receipts
-                .into_iter()
-                .map(|receipt| alloy_transaction_receipt_to_web3_transaction_receipt(receipt))
-                .collect(),
-        })
+        fetch_receipts_with_retry(alloy, hashes, block_hash, logger, supports_block_receipts)
+            .await
+            .map(|transaction_receipts| EthereumBlock {
+                block: Arc::new(alloy_block_to_web3_block(block)),
+                transaction_receipts: transaction_receipts
+                    .into_iter()
+                    .map(|receipt| alloy_transaction_receipt_to_web3_transaction_receipt(receipt))
+                    .collect(),
+            })
     }
 
     async fn get_balance(
@@ -2447,7 +2429,7 @@ fn resolve_transaction_receipt(
                 // considers this block to be in the main chain. Nothing we can do from here except
                 // give up trying to ingest this block. There is no way to get the transaction
                 // receipt from this block.
-                Err(IngestorError::BlockUnavailable(b256_to_h256(block_hash)))
+                Err(IngestorError::BlockUnavailable(block_hash))
             } else {
                 Ok(receipt)
             }

chain/ethereum/src/ingestor.rs

@@ -3,14 +3,15 @@ use crate::{EthereumAdapter, EthereumAdapterTrait as _};
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::BlockchainKind;
 use graph::components::network_provider::ChainName;
+use graph::prelude::alloy::primitives::B256;
 use graph::slog::o;
 use graph::util::backoff::ExponentialBackoff;
 use graph::{
     blockchain::{BlockHash, BlockIngestor, BlockPtr, IngestorError},
     cheap_clone::CheapClone,
     prelude::{
-        async_trait, error, info, tokio, trace, warn, web3::types::H256, ChainStore, Error,
-        EthereumBlockWithCalls, LogCode, Logger,
+        async_trait, error, info, tokio, trace, warn, ChainStore, Error, EthereumBlockWithCalls,
+        LogCode, Logger,
     },
 };
 use std::{sync::Arc, time::Duration};
@@ -168,8 +169,8 @@ impl PollingBlockIngestor {
         eth_adapter: &Arc<EthereumAdapter>,
         block_hash: &BlockHash,
     ) -> Result<Option<BlockHash>, IngestorError> {
-        // TODO: H256::from_slice can panic
-        let block_hash = H256::from_slice(block_hash.as_slice());
+        // TODO: B256::from_slice can panic
+        let block_hash = B256::from_slice(block_hash.as_slice());
 
         // Get the fully populated block
         let block = eth_adapter

graph/src/blockchain/mod.rs

@@ -47,7 +47,6 @@ use std::{
     str::FromStr,
     sync::Arc,
 };
-use web3::types::H256;
 
 pub use block_stream::{ChainHeadUpdateListener, ChainHeadUpdateStream, TriggersAdapter};
 pub use builder::{BasicBlockchainBuilder, BlockchainBuilder};
@@ -225,7 +224,7 @@ pub enum IngestorError {
     /// The Ethereum node does not know about this block for some reason, probably because it
     /// disappeared in a chain reorg.
     #[error("Block data unavailable, block was likely uncled (block hash = {0:?})")]
-    BlockUnavailable(H256),
+    BlockUnavailable(B256),
 
     /// The Ethereum node does not know about this block for some reason, probably because it
     /// disappeared in a chain reorg.

graph/src/blockchain/types.rs

@@ -97,7 +97,7 @@ impl From<Vec<u8>> for BlockHash {
 
 impl From<alloy::primitives::B256> for BlockHash {
     fn from(hash: alloy::primitives::B256) -> Self {
-        BlockHash(hash.0.to_vec().into())
+        BlockHash(hash.as_slice().into())
     }
 }
 

graph/src/util/conversions.rs

@@ -95,7 +95,13 @@ pub fn alloy_transaction_receipt_to_web3_transaction_receipt(
 }
 
 /// Converts alloy Block to web3 Block
-pub fn alloy_block_to_web3_block(_block: Arc<AlloyBlock>) -> Arc<Web3Block<Web3Transaction>> {
+pub fn alloy_block_to_web3_block_arc(_block: Arc<AlloyBlock>) -> Arc<Web3Block<Web3Transaction>> {
+    unimplemented!(
+        "Block conversion from alloy to web3 not yet implemented - will be done when needed"
+    )
+}
+
+pub fn alloy_block_to_web3_block(_block: AlloyBlock) -> Web3Block<Web3Transaction> {
     unimplemented!(
         "Block conversion from alloy to web3 not yet implemented - will be done when needed"
     )

node/src/manager/commands/chain.rs

@@ -12,9 +12,9 @@ use graph::components::store::ChainIdStore;
 use graph::components::store::StoreError;
 use graph::prelude::BlockNumber;
 use graph::prelude::ChainStore as _;
-use graph::prelude::LightEthereumBlockExt;
 use graph::prelude::{anyhow, anyhow::bail};
 use graph::slog::Logger;
+use graph::util::conversions::alloy_block_to_web3_block;
 use graph::{
     components::store::BlockStore as _, components::store::ChainHeadStore as _,
     prelude::anyhow::Error,
@@ -276,14 +276,18 @@ pub async fn ingest(
     else {
         bail!("block number {number} not found");
     };
-    let ptr = block.block_ptr();
+    let hash = block.header.hash;
+    let number = block.header.number;
     // For inserting the block, it doesn't matter whether the block is final or not.
-    let block = Arc::new(BlockFinality::Final(Arc::new(block)));
+    let block = Arc::new(BlockFinality::Final(Arc::new(alloy_block_to_web3_block(
+        block,
+    ))));
     chain_store.upsert_block(block).await?;
 
-    let rows = chain_store.confirm_block_hash(ptr.number, &ptr.hash)?;
+    let hash = hash.into();
+    let rows = chain_store.confirm_block_hash(number as i32, &hash)?;
 
-    println!("Inserted block {}", ptr);
+    println!("Inserted block {}", hash);
     if rows > 0 {
         println!("    (also deleted {rows} duplicate row(s) with that number)");
     }

node/src/manager/commands/check_blocks.rs

@@ -155,7 +155,10 @@ mod steps {
 
     use graph::{
         anyhow::bail,
-        prelude::serde_json::{self, Value},
+        prelude::{
+            h256_to_b256,
+            serde_json::{self, Value},
+        },
     };
     use json_structural_diff::{colorize as diff_to_string, JsonDiff};
 
@@ -201,13 +204,14 @@ mod steps {
         ethereum_adapter: &EthereumAdapter,
         logger: &Logger,
     ) -> anyhow::Result<Value> {
+        let block_hash = h256_to_b256(*block_hash);
         let provider_block = ethereum_adapter
-            .block_by_hash(logger, *block_hash)
+            .block_by_hash(logger, block_hash)
             .await
             .with_context(|| format!("failed to fetch block {block_hash}"))?
             .ok_or_else(|| anyhow!("JRPC provider found no block with hash {block_hash:?}"))?;
         ensure!(
-            provider_block.hash == Some(*block_hash),
+            provider_block.header.hash == block_hash,
             "Provider responded with a different block hash"
         );
         serde_json::to_value(provider_block)