chain/ethereum: migrate get_balance call to use alloy

Author: incrypto32

chain/ethereum/src/adapter.rs

@@ -1,3 +1,5 @@
+use alloy::transports::RpcError;
+use alloy::transports::TransportErrorKind;
 use anyhow::Error;
 use graph::abi;
 use graph::blockchain::ChainIdentifier;
@@ -9,7 +11,6 @@ use graph::firehose::CombinedFilter;
 use graph::firehose::LogFilter;
 use graph::prelude::web3::types::Bytes;
 use graph::prelude::web3::types::H160;
-use graph::prelude::web3::types::U256;
 use itertools::Itertools;
 use prost::Message;
 use prost_types::Any;
@@ -95,6 +96,8 @@ impl EventSignatureWithTopics {
 pub enum EthereumRpcError {
     #[error("call error: {0}")]
     Web3Error(web3::Error),
+    #[error("call error: {0}")]
+    AlloyError(RpcError<TransportErrorKind>),
     #[error("ethereum node took too long to perform call")]
     Timeout,
 }
@@ -1171,9 +1174,9 @@ pub trait EthereumAdapter: Send + Sync + 'static {
     async fn get_balance(
         &self,
         logger: &Logger,
-        address: H160,
+        address: alloy::primitives::Address,
         block_ptr: BlockPtr,
-    ) -> Result<U256, EthereumRpcError>;
+    ) -> Result<alloy::primitives::U256, EthereumRpcError>;
 
     // Returns the compiled bytecode of a smart contract
     async fn get_code(

chain/ethereum/src/ethereum_adapter.rs

@@ -22,7 +22,6 @@ use graph::futures03::{
     self, compat::Future01CompatExt, FutureExt, StreamExt, TryFutureExt, TryStreamExt,
 };
 use graph::prelude::tokio::try_join;
-use graph::prelude::web3::types::U256;
 use graph::slog::o;
 use graph::tokio::sync::RwLock;
 use graph::tokio::time::timeout;
@@ -262,7 +261,7 @@ impl EthereumAdapter {
                 client: _,
                 metrics: _,
                 provider: _,
-                rpc_url,
+                url: rpc_url,
             } => rpc_url.clone(),
             Transport::IPC(_ipc) => todo!(),
             Transport::WS(_web_socket) => todo!(),
@@ -649,6 +648,14 @@ impl EthereumAdapter {
         }
     }
 
+    fn block_ptr_to_alloy_block_id(&self, block_ptr: &BlockPtr) -> alloy_rpc_types::BlockId {
+        if !self.supports_eip_1898 {
+            alloy_rpc_types::BlockId::number(block_ptr.number as u64)
+        } else {
+            alloy_rpc_types::BlockId::hash(B256::new(*block_ptr.hash_as_h256().as_fixed_bytes()))
+        }
+    }
+
     async fn code(
         &self,
         logger: &Logger,
@@ -687,13 +694,13 @@ impl EthereumAdapter {
     async fn balance(
         &self,
         logger: &Logger,
-        address: Address,
+        address: alloy::primitives::Address,
         block_ptr: BlockPtr,
-    ) -> Result<U256, EthereumRpcError> {
-        let web3 = self.web3.clone();
+    ) -> Result<alloy::primitives::U256, EthereumRpcError> {
+        let alloy = self.alloy.clone();
         let logger = Logger::new(&logger, o!("provider" => self.provider.clone()));
 
-        let block_id = self.block_ptr_to_id(&block_ptr);
+        let block_id = self.block_ptr_to_alloy_block_id(&block_ptr);
         let retry_log_message = format!("eth_getBalance RPC call for block {}", block_ptr);
 
         retry(retry_log_message, &logger)
@@ -705,13 +712,12 @@ impl 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 {
-                    let result: Result<U256, web3::Error> =
-                        web3.eth().balance(address, Some(block_id)).boxed().await;
+                    let result = alloy.get_balance(address).block_id(block_id).await;
                     match result {
                         Ok(balance) => Ok(balance),
-                        Err(err) => Err(EthereumRpcError::Web3Error(err)),
+                        Err(err) => Err(EthereumRpcError::AlloyError(err)),
                     }
                 }
             })
@@ -1508,9 +1514,9 @@ impl EthereumAdapterTrait for EthereumAdapter {
     async fn get_balance(
         &self,
         logger: &Logger,
-        address: H160,
+        address: alloy::primitives::Address,
         block_ptr: BlockPtr,
-    ) -> Result<U256, EthereumRpcError> {
+    ) -> Result<alloy::primitives::U256, EthereumRpcError> {
         debug!(
             logger, "eth_getBalance";
             "address" => format!("{}", address),

chain/ethereum/src/runtime/runtime_adapter.rs

@@ -224,17 +224,18 @@ fn eth_get_balance(
     let logger = &ctx.logger;
     let block_ptr = &ctx.block_ptr;
 
-    let address: H160 = asc_get(ctx.heap, wasm_ptr.into(), &ctx.gas, 0)?;
+    let address: alloy::primitives::Address = asc_get(ctx.heap, wasm_ptr.into(), &ctx.gas, 0)?;
 
     let result = graph::block_on(eth_adapter.get_balance(logger, address, block_ptr.clone()));
 
     match result {
         Ok(v) => {
-            let bigint = BigInt::from_unsigned_u256(&v);
+            let bigint = BigInt::from_unsigned_alloy_u256(&v);
             Ok(asc_new(ctx.heap, &bigint, &ctx.gas)?)
         }
         // Retry on any kind of error
         Err(EthereumRpcError::Web3Error(e)) => Err(HostExportError::PossibleReorg(e.into())),
+        Err(EthereumRpcError::AlloyError(e)) => Err(HostExportError::PossibleReorg(e.into())),
         Err(EthereumRpcError::Timeout) => Err(HostExportError::PossibleReorg(
             EthereumRpcError::Timeout.into(),
         )),
@@ -267,6 +268,7 @@ fn eth_has_code(
         Ok(v) => Ok(asc_new(ctx.heap, &AscWrapped { inner: v }, &ctx.gas)?),
         // Retry on any kind of error
         Err(EthereumRpcError::Web3Error(e)) => Err(HostExportError::PossibleReorg(e.into())),
+        Err(EthereumRpcError::AlloyError(e)) => Err(HostExportError::PossibleReorg(e.into())),
         Err(EthereumRpcError::Timeout) => Err(HostExportError::PossibleReorg(
             EthereumRpcError::Timeout.into(),
         )),

graph/src/data/store/scalar/bigint.rs

@@ -186,6 +186,12 @@ impl BigInt {
         BigInt::from_unsigned_bytes_le(&bytes).unwrap()
     }
 
+    pub fn from_unsigned_alloy_u256(n: &alloy::primitives::U256) -> Self {
+        let bytes = n.to_le_bytes_trimmed_vec();
+        // Unwrap: 256 bits is much less than BigInt::MAX_BITS
+        BigInt::from_unsigned_bytes_le(&bytes).unwrap()
+    }
+
     pub fn from_signed_u256(n: &U256) -> Self {
         let mut bytes: [u8; 32] = [0; 32];
         n.to_little_endian(&mut bytes);

runtime/wasm/Cargo.toml

@@ -4,6 +4,7 @@ version.workspace = true
 edition.workspace = true
 
 [dependencies]
+alloy = { workspace = true }
 async-trait = "0.1.50"
 hex = "0.4.3"
 graph = { path = "../../graph" }

runtime/wasm/src/to_from/external.rs

@@ -46,6 +46,28 @@ impl FromAscObj<Uint8Array> for web3::H160 {
     }
 }
 
+impl FromAscObj<Uint8Array> for alloy::primitives::Address {
+    fn from_asc_obj<H: AscHeap + ?Sized>(
+        typed_array: Uint8Array,
+        heap: &H,
+        gas: &GasCounter,
+        depth: usize,
+    ) -> Result<Self, DeterministicHostError> {
+        let data = <[u8; 20]>::from_asc_obj(typed_array, heap, gas, depth)?;
+        Ok(Self::from(data))
+    }
+}
+
+impl ToAscObj<Uint8Array> for alloy::primitives::Address {
+    fn to_asc_obj<H: AscHeap + ?Sized>(
+        &self,
+        heap: &mut H,
+        gas: &GasCounter,
+    ) -> Result<Uint8Array, HostExportError> {
+        self.as_slice().to_asc_obj(heap, gas)
+    }
+}
+
 impl FromAscObj<Uint8Array> for web3::H256 {
     fn from_asc_obj<H: AscHeap + ?Sized>(
         typed_array: Uint8Array,