chain, graph: Fix block ptr ext parsing bug

chain/ethereum/src/ethereum_adapter.rs

@@ -1707,7 +1707,7 @@ impl EthereumAdapterTrait for EthereumAdapter {
         chain_store: Arc<dyn ChainStore>,
         block_numbers: HashSet<BlockNumber>,
     ) -> Box<dyn Stream<Item = Arc<BlockPtrExt>, Error = Error> + Send> {
-        let blocks_map: BTreeMap<i32, Vec<json::Value>> = chain_store
+        let blocks_map = chain_store
             .cheap_clone()
             .block_ptrs_by_numbers(block_numbers.iter().map(|&b| b.into()).collect::<Vec<_>>())
             .await
@@ -1721,7 +1721,7 @@ impl EthereumAdapterTrait for EthereumAdapter {
             .into_iter()
             .filter_map(|(_number, values)| {
                 if values.len() == 1 {
-                    json::from_value(values[0].clone()).ok()
+                    Arc::new(values[0].clone()).into()
                 } else {
                     None
                 }

graph/src/blockchain/types.rs

@@ -482,6 +482,21 @@ impl TryFrom<(Option<H256>, Option<U64>, H256, U256)> for BlockPtrExt {
         })
     }
 }
+
+impl TryFrom<(H256, i32, H256, U256)> for BlockPtrExt {
+    type Error = anyhow::Error;
+
+    fn try_from(tuple: (H256, i32, H256, U256)) -> Result<Self, Self::Error> {
+        let (hash, block_number, parent_hash, timestamp) = tuple;
+
+        Ok(BlockPtrExt {
+            hash: hash.into(),
+            number: block_number,
+            parent_hash: parent_hash.into(),
+            timestamp,
+        })
+    }
+}
 impl From<BlockPtrExt> for H256 {
     fn from(ptr: BlockPtrExt) -> Self {
         ptr.hash_as_h256()

graph/src/components/store/traits.rs

@@ -7,7 +7,7 @@ use web3::types::{Address, H256};
 
 use super::*;
 use crate::blockchain::block_stream::{EntityWithType, FirehoseCursor};
-use crate::blockchain::{BlockTime, ChainIdentifier};
+use crate::blockchain::{BlockPtrExt, BlockTime, ChainIdentifier};
 use crate::components::metrics::stopwatch::StopwatchMetrics;
 use crate::components::server::index_node::VersionInfo;
 use crate::components::subgraph::SubgraphVersionSwitchingMode;
@@ -500,7 +500,7 @@ pub trait ChainStore: Send + Sync + 'static {
     async fn block_ptrs_by_numbers(
         self: Arc<Self>,
         numbers: Vec<BlockNumber>,
-    ) -> Result<BTreeMap<BlockNumber, Vec<serde_json::Value>>, Error>;
+    ) -> Result<BTreeMap<BlockNumber, Vec<BlockPtrExt>>, Error>;
 
     /// Get the `offset`th ancestor of `block_hash`, where offset=0 means the block matching
     /// `block_hash` and offset=1 means its parent. If `root` is passed, short-circuit upon finding

store/postgres/src/chain_store.rs

@@ -21,9 +21,9 @@ use std::{
     sync::Arc,
 };
 
-use graph::blockchain::{Block, BlockHash, ChainIdentifier};
+use graph::blockchain::{Block, BlockHash, BlockPtrExt, ChainIdentifier};
 use graph::cheap_clone::CheapClone;
-use graph::prelude::web3::types::H256;
+use graph::prelude::web3::types::{H256, U256};
 use graph::prelude::{
     async_trait, serde_json as json, transaction_receipt::LightTransactionReceipt, BlockNumber,
     BlockPtr, CachedEthereumCall, CancelableError, ChainStore as ChainStoreTrait, Error,
@@ -53,6 +53,14 @@ impl JsonBlock {
             data,
         }
     }
+
+    fn timestamp(&self) -> Option<U256> {
+        self.data
+            .as_ref()
+            .and_then(|data| data.get("timestamp"))
+            .and_then(|ts| ts.as_str())
+            .and_then(|ts| U256::from_dec_str(ts).ok())
+    }
 }
 
 /// Tables in the 'public' database schema that store chain-specific data
@@ -1949,6 +1957,16 @@ impl ChainStore {
     }
 }
 
+fn json_block_to_block_ptr_ext(json_block: &JsonBlock) -> Result<BlockPtrExt, Error> {
+    let hash = json_block.ptr.hash.clone();
+    let number = json_block.ptr.number;
+    let parent_hash = json_block.parent_hash.clone();
+    let timestamp = json_block.timestamp().unwrap();
+    let ptr = BlockPtrExt::try_from((hash.as_h256(), number, parent_hash.as_h256(), timestamp))?;
+
+    Ok(ptr)
+}
+
 #[async_trait]
 impl ChainStoreTrait for ChainStore {
     fn genesis_block_ptr(&self) -> Result<BlockPtr, Error> {
@@ -2145,23 +2163,11 @@ impl ChainStoreTrait for ChainStore {
     async fn block_ptrs_by_numbers(
         self: Arc<Self>,
         numbers: Vec<BlockNumber>,
-    ) -> Result<BTreeMap<BlockNumber, Vec<json::Value>>, Error> {
-        if ENV_VARS.store.disable_block_cache_for_lookup {
-            let values = self
-                .blocks_from_store_by_numbers(numbers)
-                .await?
-                .into_iter()
-                .map(|(num, blocks)| {
-                    (
-                        num,
-                        blocks
-                            .into_iter()
-                            .filter_map(|block| block.data)
-                            .collect::<Vec<_>>(),
-                    )
-                })
-                .collect();
-            Ok(values)
+    ) -> Result<BTreeMap<BlockNumber, Vec<BlockPtrExt>>, Error> {
+        let result = if ENV_VARS.store.disable_block_cache_for_lookup {
+            let values = self.blocks_from_store_by_numbers(numbers).await?;
+
+            values
         } else {
             let cached = self.recent_blocks_cache.get_block_ptrs_by_numbers(&numbers);
 
@@ -2209,16 +2215,28 @@ impl ChainStoreTrait for ChainStore {
                 .map(|(ptr, data)| (ptr.block_number(), vec![data]))
                 .collect::<BTreeMap<_, _>>();
 
-            let mut result: BTreeMap<BlockNumber, Vec<json::Value>> = cached_map;
+            let mut result = cached_map;
             for (num, blocks) in stored {
-                result
-                    .entry(num)
-                    .or_default()
-                    .extend(blocks.into_iter().filter_map(|block| block.data));
+                if !result.contains_key(&num) {
+                    result.insert(num, blocks);
+                }
             }
 
-            Ok(result)
-        }
+            result
+        };
+
+        let ptrs = result
+            .into_iter()
+            .map(|(num, blocks)| {
+                let ptrs = blocks
+                    .into_iter()
+                    .filter_map(|block| json_block_to_block_ptr_ext(&block).ok())
+                    .collect();
+                (num, ptrs)
+            })
+            .collect();
+
+        Ok(ptrs)
     }
 
     async fn blocks(self: Arc<Self>, hashes: Vec<BlockHash>) -> Result<Vec<json::Value>, Error> {
@@ -2520,10 +2538,8 @@ mod recent_blocks_cache {
                 .and_then(|block| block.data.as_ref().map(|data| (&block.ptr, data)))
         }
 
-        fn get_block_by_number(&self, number: BlockNumber) -> Option<(&BlockPtr, &json::Value)> {
-            self.blocks
-                .get(&number)
-                .and_then(|block| block.data.as_ref().map(|data| (&block.ptr, data)))
+        fn get_block_by_number(&self, number: BlockNumber) -> Option<&JsonBlock> {
+            self.blocks.get(&number)
         }
 
         fn get_ancestor(
@@ -2651,13 +2667,13 @@ mod recent_blocks_cache {
         pub fn get_block_ptrs_by_numbers(
             &self,
             numbers: &[BlockNumber],
-        ) -> Vec<(BlockPtr, json::Value)> {
+        ) -> Vec<(BlockPtr, JsonBlock)> {
             let inner = self.inner.read();
-            let mut blocks: Vec<(BlockPtr, json::Value)> = Vec::new();
+            let mut blocks: Vec<(BlockPtr, JsonBlock)> = Vec::new();
 
             for &number in numbers {
-                if let Some((ptr, block)) = inner.get_block_by_number(number) {
-                    blocks.push((ptr.clone(), block.clone()));
+                if let Some(block) = inner.get_block_by_number(number) {
+                    blocks.push((block.ptr.clone(), block.clone()));
                 }
             }