feat: implement eth_getBalance, eth_getCode and eth_getStorageAt

rpc/Cargo.toml

@@ -12,7 +12,9 @@ authors = ["https://github.com/ethereum/trin/graphs/contributors"]
 
 [dependencies]
 alloy-primitives = "0.7.0"
+alloy-rlp = "0.3.4"
 discv5 = { version = "0.4.1", features = ["serde"] }
+eth_trie = { git = "https://github.com/kolbyml/eth-trie.rs.git", rev = "7947a83091192a7988f359b750b05121d5d7ba8c" }
 ethportal-api = { path = "../ethportal-api"}
 portalnet = { path = "../portalnet"}
 tracing = "0.1.27"

rpc/src/errors.rs

@@ -53,7 +53,6 @@ impl RpcError {
     }
 }
 
-#[allow(clippy::large_enum_variant)]
 pub enum RpcServeError {
     /// A generic error with no data
     Message(String),
@@ -62,7 +61,7 @@ pub enum RpcServeError {
     /// ContentNotFound
     ContentNotFound {
         message: String,
-        trace: Option<QueryTrace>,
+        trace: Option<Box<QueryTrace>>,
     },
 }
 
@@ -94,7 +93,7 @@ impl From<ContentNotFoundJsonError> for RpcServeError {
     fn from(e: ContentNotFoundJsonError) -> Self {
         RpcServeError::ContentNotFound {
             message: e.message,
-            trace: e.trace,
+            trace: e.trace.map(Box::new),
         }
     }
 }

rpc/src/eth_rpc.rs

@@ -1,25 +1,38 @@
-use alloy_primitives::{Address, Bytes, B256, U256};
+use alloy_primitives::{keccak256, Address, Bytes, B256, U256};
+use alloy_rlp::Decodable;
+use eth_trie::node::Node;
 use reth_rpc_types::{other::OtherFields, Block, BlockId, BlockTransactions};
 use tokio::sync::mpsc;
 
 use ethportal_api::{
     types::{
+        content_key::state::{AccountTrieNodeKey, ContractBytecodeKey, ContractStorageTrieNodeKey},
         execution::block_body::BlockBody,
-        jsonrpc::request::{HistoryJsonRpcRequest, StateJsonRpcRequest},
+        jsonrpc::{
+            endpoints::{HistoryEndpoint, StateEndpoint},
+            request::{HistoryJsonRpcRequest, StateJsonRpcRequest},
+        },
+        portal::ContentInfo,
+        state_trie::{
+            account_state::AccountState,
+            nibbles::Nibbles,
+            trie_traversal::{NodeTraversal, TraversalResult},
+        },
     },
-    EthApiServer,
+    ContentValue, EthApiServer, Header, HistoryContentKey, HistoryContentValue, StateContentKey,
+    StateContentValue,
 };
 use trin_validation::constants::CHAIN_ID;
 
 use crate::{
     errors::RpcServeError,
-    fetch::{find_block_body_by_hash, find_header_by_hash},
+    fetch::proxy_to_subnet,
     jsonrpsee::core::{async_trait, RpcResult},
 };
 
 pub struct EthApi {
     history_network: mpsc::UnboundedSender<HistoryJsonRpcRequest>,
-    _state_network: Option<mpsc::UnboundedSender<StateJsonRpcRequest>>,
+    state_network: Option<mpsc::UnboundedSender<StateJsonRpcRequest>>,
 }
 
 impl EthApi {
@@ -29,7 +42,7 @@ impl EthApi {
     ) -> Self {
         Self {
             history_network,
-            _state_network: state_network,
+            state_network,
         }
     }
 }
@@ -52,8 +65,8 @@ impl EthApiServer for EthApi {
             .into());
         }
 
-        let header = find_header_by_hash(&self.history_network, block_hash).await?;
-        let body = find_block_body_by_hash(&self.history_network, block_hash).await?;
+        let header = self.fetch_header_by_hash(block_hash).await?;
+        let body = self.fetch_block_body(block_hash).await?;
         let transactions = match body {
             BlockBody::Legacy(body) => body.txs,
             BlockBody::Merge(body) => body.txs,
@@ -78,21 +91,235 @@ impl EthApiServer for EthApi {
         Ok(block)
     }
 
-    async fn get_balance(&self, _address: Address, _block: BlockId) -> RpcResult<U256> {
-        todo!()
+    async fn get_balance(&self, address: Address, block: BlockId) -> RpcResult<U256> {
+        let address_hash = keccak256(address);
+        let block_hash = as_block_hash(block)?;
+        let header = self.fetch_header_by_hash(block_hash).await?;
+        match self
+            .fetch_account_state(header.state_root, address_hash)
+            .await?
+        {
+            Some(account_state) => Ok(account_state.balance),
+            None => Ok(U256::ZERO),
+        }
     }
 
-    async fn get_code(&self, _address: Address, _block: BlockId) -> RpcResult<Bytes> {
-        todo!()
+    async fn get_code(&self, address: Address, block: BlockId) -> RpcResult<Bytes> {
+        let address_hash = keccak256(address);
+        let block_hash = as_block_hash(block)?;
+        let header = self.fetch_header_by_hash(block_hash).await?;
+        let account_state = self
+            .fetch_account_state(header.state_root, address_hash)
+            .await?;
+        let Some(account_state) = account_state else {
+            return Ok(Bytes::new());
+        };
+
+        Ok(self
+            .fetch_contract_bytecode(address_hash, account_state.code_hash)
+            .await?)
     }
 
     async fn get_storage_at(
         &self,
-        _address: Address,
-        _slot: U256,
-        _block: BlockId,
+        address: Address,
+        slot: U256,
+        block: BlockId,
     ) -> RpcResult<Bytes> {
-        todo!()
+        let address_hash = keccak256(address);
+        let block_hash = as_block_hash(block)?;
+        let header = self.fetch_header_by_hash(block_hash).await?;
+        let account_state = self
+            .fetch_account_state(header.state_root, address_hash)
+            .await?;
+        let Some(account_state) = account_state else {
+            return Ok(Bytes::new());
+        };
+
+        Ok(self
+            .fetch_contract_storage_at_slot(account_state.storage_root, address_hash, slot)
+            .await?)
+    }
+}
+
+impl EthApi {
+    // History network related functions
+
+    async fn fetch_history_content(
+        &self,
+        content_key: HistoryContentKey,
+    ) -> Result<HistoryContentValue, RpcServeError> {
+        let endpoint = HistoryEndpoint::RecursiveFindContent(content_key.clone());
+        let response: ContentInfo = proxy_to_subnet(&self.history_network, endpoint).await?;
+        let ContentInfo::Content { content, .. } = response else {
+            return Err(RpcServeError::Message(format!(
+                "Invalid response variant: History RecursiveFindContent should contain content value; got {response:?}"
+            )));
+        };
+
+        let content_value = HistoryContentValue::decode(&content_key, &content)?;
+        Ok(content_value)
+    }
+
+    async fn fetch_header_by_hash(&self, block_hash: B256) -> Result<Header, RpcServeError> {
+        let content_value = self
+            .fetch_history_content(HistoryContentKey::BlockHeaderWithProof(block_hash.into()))
+            .await?;
+        let HistoryContentValue::BlockHeaderWithProof(header_with_proof) = content_value else {
+            return Err(RpcServeError::Message(format!(
+                "Invalid response: expected block header; got {content_value:?}"
+            )));
+        };
+        Ok(header_with_proof.header)
+    }
+
+    async fn fetch_block_body(&self, block_hash: B256) -> Result<BlockBody, RpcServeError> {
+        let content_value = self
+            .fetch_history_content(HistoryContentKey::BlockBody(block_hash.into()))
+            .await?;
+        let HistoryContentValue::BlockBody(block_body) = content_value else {
+            return Err(RpcServeError::Message(format!(
+                "Invalid response: expected block body; got {content_value:?}"
+            )));
+        };
+        Ok(block_body)
+    }
+
+    // State network related functions
+
+    async fn fetch_state_content(
+        &self,
+        content_key: StateContentKey,
+    ) -> Result<StateContentValue, RpcServeError> {
+        let Some(state_network) = &self.state_network else {
+            return Err(RpcServeError::Message(format!(
+                "State network not enabled. Can't find: {content_key}"
+            )));
+        };
+
+        let endpoint = StateEndpoint::RecursiveFindContent(content_key.clone());
+        let response: ContentInfo = proxy_to_subnet(state_network, endpoint).await?;
+        let ContentInfo::Content { content, .. } = response else {
+            return Err(RpcServeError::Message(format!(
+                "Invalid response variant: State RecursiveFindContent should contain content value; got {response:?}"
+            )));
+        };
+
+        let content_value = StateContentValue::decode(&content_key, &content)?;
+        Ok(content_value)
+    }
+
+    async fn fetch_trie_node(&self, content_key: StateContentKey) -> Result<Node, RpcServeError> {
+        let content_value = self.fetch_state_content(content_key).await?;
+        let StateContentValue::TrieNode(trie_node) = content_value else {
+            return Err(RpcServeError::Message(format!(
+                "Invalid response: expected trie node; got {content_value:?}",
+            )));
+        };
+        trie_node
+            .node
+            .as_trie_node()
+            .map_err(|err| RpcServeError::Message(format!("Can't decode trie_node: {err}")))
+    }
+
+    async fn fetch_contract_bytecode(
+        &self,
+        address_hash: B256,
+        code_hash: B256,
+    ) -> Result<Bytes, RpcServeError> {
+        let content_key = StateContentKey::ContractBytecode(ContractBytecodeKey {
+            address_hash,
+            code_hash,
+        });
+        let content_value = self.fetch_state_content(content_key).await?;
+        let StateContentValue::ContractBytecode(contract_bytecode) = content_value else {
+            return Err(RpcServeError::Message(format!(
+                "Invalid response: expected contract bytecode; got {content_value:?}",
+            )));
+        };
+        let bytes = Vec::from(contract_bytecode.code);
+        Ok(Bytes::from(bytes))
+    }
+
+    async fn fetch_account_state(
+        &self,
+        state_root: B256,
+        address_hash: B256,
+    ) -> Result<Option<AccountState>, RpcServeError> {
+        let value = self
+            .traverse_trie(state_root, address_hash, |path, node_hash| {
+                StateContentKey::AccountTrieNode(AccountTrieNodeKey { path, node_hash })
+            })
+            .await?;
+        let Some(account_state_bytes) = value else {
+            return Ok(None);
+        };
+        let account_state = AccountState::decode(&mut account_state_bytes.as_slice())
+            .map_err(|err| RpcServeError::Message(format!("Invalid account state data: {err}")))?;
+        Ok(Some(account_state))
+    }
+
+    async fn fetch_contract_storage_at_slot(
+        &self,
+        storage_root: B256,
+        address_hash: B256,
+        storage_slot: U256,
+    ) -> Result<Bytes, RpcServeError> {
+        let path = keccak256(storage_slot.to_be_bytes::<32>());
+        let value = self
+            .traverse_trie(storage_root, path, |path, node_hash| {
+                StateContentKey::ContractStorageTrieNode(ContractStorageTrieNodeKey {
+                    address_hash,
+                    path,
+                    node_hash,
+                })
+            })
+            .await?;
+        Ok(Bytes::from(value.unwrap_or_default()))
+    }
+
+    /// Utility function for fetching trie nodes and traversing the trie.
+    ///
+    /// This function works both with the account trie and the contract state trie.
+    async fn traverse_trie(
+        &self,
+        root: B256,
+        path: B256,
+        content_key_fn: impl Fn(Nibbles, B256) -> StateContentKey,
+    ) -> Result<Option<Vec<u8>>, RpcServeError> {
+        let path = Nibbles::unpack_nibbles(path.as_slice());
+
+        let mut node_hash = root;
+        let mut remaining_path = path.as_slice();
+
+        loop {
+            let path_from_root = path
+                .strip_suffix(remaining_path)
+                .expect("Remaining path should be suffix of the path");
+
+            let content_key = content_key_fn(
+                Nibbles::try_from_unpacked_nibbles(path_from_root)
+                    .expect("we should be able to create Nibbles from path"),
+                node_hash,
+            );
+            let node = self.fetch_trie_node(content_key).await?;
+
+            match node.traverse(remaining_path) {
+                TraversalResult::Empty(_) => return Ok(None),
+                TraversalResult::Value(value) => {
+                    return Ok(Some(value));
+                }
+                TraversalResult::Node(next_node) => {
+                    node_hash = next_node.hash;
+                    remaining_path = next_node.remaining_path;
+                }
+                TraversalResult::Error(err) => {
+                    return Err(RpcServeError::Message(format!(
+                        "Error traversing trie node: {err}"
+                    )))
+                }
+            }
+        }
     }
 }
 
@@ -101,3 +328,9 @@ impl std::fmt::Debug for EthApi {
         f.debug_struct("EthApi").finish_non_exhaustive()
     }
 }
+
+fn as_block_hash(block: BlockId) -> Result<B256, RpcServeError> {
+    block.as_block_hash().ok_or_else(|| {
+        RpcServeError::Message("Only block hash is accepted as block id".to_string())
+    })
+}