graph, chain/ethereum: use block ptrs when fetching blocks for composed subgraphs

Author: incrypto32

chain/arweave/src/chain.rs

@@ -269,7 +269,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
         }))
     }
 
-    async fn load_blocks_by_numbers(
+    async fn load_block_ptrs_by_numbers(
         &self,
         _logger: Logger,
         _block_numbers: HashSet<BlockNumber>,

chain/cosmos/src/chain.rs

@@ -197,15 +197,13 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
     ) -> Result<Option<codec::Block>, Error> {
         panic!("Should never be called since not used by FirehoseBlockStream")
     }
-
-    async fn load_blocks_by_numbers(
+    async fn load_block_ptrs_by_numbers(
         &self,
         _logger: Logger,
         _block_numbers: HashSet<BlockNumber>,
     ) -> Result<Vec<Block>, Error> {
-        unimplemented!()
+        todo!()
     }
-
     async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
         unimplemented!()
     }

chain/ethereum/src/adapter.rs

@@ -1,5 +1,6 @@
 use anyhow::Error;
 use ethabi::{Error as ABIError, Function, ParamType, Token};
+use graph::blockchain::BlockPtrExt;
 use graph::blockchain::ChainIdentifier;
 use graph::components::subgraph::MappingError;
 use graph::data::store::ethereum::call;
@@ -1109,12 +1110,12 @@ pub trait EthereumAdapter: Send + Sync + 'static {
         block_hash: H256,
     ) -> Box<dyn Future<Item = LightEthereumBlock, Error = Error> + Send>;
 
-    async fn load_blocks_by_numbers(
+    async fn load_block_ptrs_by_numbers(
         &self,
         _logger: Logger,
         _chain_store: Arc<dyn ChainStore>,
         _block_numbers: HashSet<BlockNumber>,
-    ) -> Box<dyn Stream<Item = Arc<LightEthereumBlock>, Error = Error> + Send>;
+    ) -> Box<dyn Stream<Item = Arc<BlockPtrExt>, Error = Error> + Send>;
 
     /// Load Ethereum blocks in bulk, returning results as they come back as a Stream.
     /// May use the `chain_store` as a cache.

chain/ethereum/src/chain.rs

@@ -3,7 +3,7 @@ use anyhow::{Context, Error};
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::firehose_block_ingestor::{FirehoseBlockIngestor, Transforms};
 use graph::blockchain::{
-    BlockIngestor, BlockTime, BlockchainKind, ChainIdentifier, TriggerFilterWrapper,
+    BlockIngestor, BlockPtrExt, BlockTime, BlockchainKind, ChainIdentifier, TriggerFilterWrapper,
     TriggersAdapterSelector,
 };
 use graph::components::adapter::ChainId;
@@ -156,6 +156,7 @@ impl BlockStreamBuilder<Chain> for EthereumStreamBuilder {
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Chain>>> {
         let requirements = filter.chain_filter.node_capabilities();
+        let is_using_subgraph_composition = !source_subgraph_stores.is_empty();
         let adapter = TriggersAdapterWrapper::new(
             chain
                 .triggers_adapter(&deployment, &requirements, unified_api_version.clone())
@@ -181,20 +182,26 @@ impl BlockStreamBuilder<Chain> for EthereumStreamBuilder {
         // This is ok because Celo blocks are always final. And we _need_ to do this because
         // some events appear only in eth_getLogs but not in transaction receipts.
         // See also ca0edc58-0ec5-4c89-a7dd-2241797f5e50.
-        let chain_id = match chain.chain_client().as_ref() {
+        let reorg_threshold = match chain.chain_client().as_ref() {
             ChainClient::Rpc(adapter) => {
-                adapter
+                let chain_id = adapter
                     .cheapest()
                     .await
                     .ok_or(anyhow!("unable to get eth adapter for chan_id call"))?
                     .chain_id()
-                    .await?
+                    .await?;
+
+                if CELO_CHAIN_IDS.contains(&chain_id) {
+                    0
+                } else {
+                    chain.reorg_threshold
+                }
             }
-            _ => panic!("expected rpc when using polling blockstream"),
-        };
-        let reorg_threshold = match CELO_CHAIN_IDS.contains(&chain_id) {
-            false => chain.reorg_threshold,
-            true => 0,
+            _ if is_using_subgraph_composition => chain.reorg_threshold,
+            _ => panic!(
+                "expected rpc when using polling blockstream : {}",
+                is_using_subgraph_composition
+            ),
         };
 
         Ok(Box::new(PollingBlockStream::new(
@@ -617,6 +624,8 @@ pub enum BlockFinality {
 
     // If a block may still be reorged, we need to work with more local data.
     NonFinal(EthereumBlockWithCalls),
+
+    Ptr(Arc<BlockPtrExt>),
 }
 
 impl Default for BlockFinality {
@@ -630,6 +639,7 @@ impl BlockFinality {
         match self {
             BlockFinality::Final(block) => block,
             BlockFinality::NonFinal(block) => &block.ethereum_block.block,
+            BlockFinality::Ptr(_) => unreachable!("light_block called on HeaderOnly"),
         }
     }
 }
@@ -639,6 +649,7 @@ impl<'a> From<&'a BlockFinality> for BlockPtr {
         match block {
             BlockFinality::Final(b) => BlockPtr::from(&**b),
             BlockFinality::NonFinal(b) => BlockPtr::from(&b.ethereum_block),
+            BlockFinality::Ptr(b) => BlockPtr::new(b.hash.clone(), b.number),
         }
     }
 }
@@ -648,13 +659,17 @@ impl Block for BlockFinality {
         match self {
             BlockFinality::Final(block) => block.block_ptr(),
             BlockFinality::NonFinal(block) => block.ethereum_block.block.block_ptr(),
+            BlockFinality::Ptr(block) => BlockPtr::new(block.hash.clone(), block.number),
         }
     }
 
     fn parent_ptr(&self) -> Option<BlockPtr> {
         match self {
             BlockFinality::Final(block) => block.parent_ptr(),
             BlockFinality::NonFinal(block) => block.ethereum_block.block.parent_ptr(),
+            BlockFinality::Ptr(block) => {
+                Some(BlockPtr::new(block.parent_hash.clone(), block.number - 1))
+            }
         }
     }
 
@@ -687,13 +702,15 @@ impl Block for BlockFinality {
                 json::to_value(eth_block)
             }
             BlockFinality::NonFinal(block) => json::to_value(&block.ethereum_block),
+            BlockFinality::Ptr(_) => Ok(json::Value::Null),
         }
     }
 
     fn timestamp(&self) -> BlockTime {
         let ts = match self {
             BlockFinality::Final(block) => block.timestamp,
             BlockFinality::NonFinal(block) => block.ethereum_block.block.timestamp,
+            BlockFinality::Ptr(block) => block.timestamp,
         };
         let ts = i64::try_from(ts.as_u64()).unwrap();
         BlockTime::since_epoch(ts, 0)
@@ -735,7 +752,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
         .await
     }
 
-    async fn load_blocks_by_numbers(
+    async fn load_block_ptrs_by_numbers(
         &self,
         logger: Logger,
         block_numbers: HashSet<BlockNumber>,
@@ -749,9 +766,9 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
             .await?;
 
         let blocks = adapter
-            .load_blocks_by_numbers(logger, self.chain_store.clone(), block_numbers)
+            .load_block_ptrs_by_numbers(logger, self.chain_store.clone(), block_numbers)
             .await
-            .map(|block| BlockFinality::Final(block))
+            .map(|block| BlockFinality::Ptr(block))
             .collect()
             .compat()
             .await?;
@@ -812,6 +829,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
                 triggers.append(&mut parse_block_triggers(&filter.block, full_block));
                 Ok(BlockWithTriggers::new(block, triggers, logger))
             }
+            BlockFinality::Ptr(_) => unreachable!("triggers_in_block called on HeaderOnly"),
         }
     }
 

chain/ethereum/src/ethereum_adapter.rs

@@ -1,6 +1,7 @@
 use futures03::{future::BoxFuture, stream::FuturesUnordered};
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::BlockHash;
+use graph::blockchain::BlockPtrExt;
 use graph::blockchain::ChainIdentifier;
 
 use graph::components::transaction_receipt::LightTransactionReceipt;
@@ -783,11 +784,11 @@ impl EthereumAdapter {
     }
 
     /// Request blocks by number through JSON-RPC.
-    fn load_blocks_by_numbers_rpc(
+    fn load_block_ptrs_by_numbers_rpc(
         &self,
         logger: Logger,
         numbers: Vec<BlockNumber>,
-    ) -> impl Stream<Item = Arc<LightEthereumBlock>, Error = Error> + Send {
+    ) -> impl Stream<Item = Arc<BlockPtrExt>, Error = Error> + Send {
         let web3 = self.web3.clone();
 
         stream::iter_ok::<_, Error>(numbers.into_iter().map(move |number| {
@@ -798,19 +799,29 @@ impl EthereumAdapter {
                 .run(move || {
                     Box::pin(
                         web3.eth()
-                            .block_with_txs(BlockId::Number(Web3BlockNumber::Number(
-                                number.into(),
-                            ))),
+                            .block(BlockId::Number(Web3BlockNumber::Number(number.into()))),
                     )
                     .compat()
                     .from_err::<Error>()
                     .and_then(move |block| {
-                        block.map(Arc::new).ok_or_else(|| {
-                            anyhow::anyhow!(
-                                "Ethereum node did not find block with number {:?}",
-                                number
-                            )
-                        })
+                        block
+                            .map(|block| {
+                                let ptr = BlockPtrExt::try_from((
+                                    block.hash,
+                                    block.number,
+                                    block.parent_hash,
+                                    block.timestamp,
+                                ))
+                                .unwrap();
+
+                                Arc::new(ptr)
+                            })
+                            .ok_or_else(|| {
+                                anyhow::anyhow!(
+                                    "Ethereum node did not find block with number {:?}",
+                                    number
+                                )
+                            })
                     })
                     .compat()
                 })
@@ -1690,23 +1701,23 @@ impl EthereumAdapterTrait for EthereumAdapter {
     }
 
     /// Load Ethereum blocks in bulk by number, returning results as they come back as a Stream.
-    async fn load_blocks_by_numbers(
+    async fn load_block_ptrs_by_numbers(
         &self,
         logger: Logger,
         chain_store: Arc<dyn ChainStore>,
         block_numbers: HashSet<BlockNumber>,
-    ) -> Box<dyn Stream<Item = Arc<LightEthereumBlock>, Error = Error> + Send> {
+    ) -> Box<dyn Stream<Item = Arc<BlockPtrExt>, Error = Error> + Send> {
         let blocks_map: BTreeMap<i32, Vec<json::Value>> = chain_store
             .cheap_clone()
-            .blocks_by_numbers(block_numbers.iter().map(|&b| b.into()).collect::<Vec<_>>())
+            .block_ptrs_by_numbers(block_numbers.iter().map(|&b| b.into()).collect::<Vec<_>>())
             .await
             .map_err(|e| {
                 error!(&logger, "Error accessing block cache {}", e);
                 e
             })
             .unwrap_or_default();
 
-        let mut blocks: Vec<Arc<LightEthereumBlock>> = blocks_map
+        let mut blocks: Vec<Arc<BlockPtrExt>> = blocks_map
             .into_iter()
             .filter_map(|(_number, values)| {
                 if values.len() == 1 {
@@ -1719,7 +1730,7 @@ impl EthereumAdapterTrait for EthereumAdapter {
 
         let missing_blocks: Vec<i32> = block_numbers
             .into_iter()
-            .filter(|&number| !blocks.iter().any(|block| block.number() == number))
+            .filter(|&number| !blocks.iter().any(|block| block.block_number() == number))
             .collect();
 
         if !missing_blocks.is_empty() {
@@ -1731,20 +1742,9 @@ impl EthereumAdapterTrait for EthereumAdapter {
         }
 
         Box::new(
-            self.load_blocks_by_numbers_rpc(logger.clone(), missing_blocks)
+            self.load_block_ptrs_by_numbers_rpc(logger.clone(), missing_blocks)
                 .collect()
                 .map(move |new_blocks| {
-                    let upsert_blocks: Vec<_> = new_blocks
-                        .iter()
-                        .map(|block| BlockFinality::Final(block.clone()))
-                        .collect();
-                    let block_refs: Vec<_> = upsert_blocks
-                        .iter()
-                        .map(|block| block as &dyn graph::blockchain::Block)
-                        .collect();
-                    if let Err(e) = chain_store.upsert_light_blocks(block_refs.as_slice()) {
-                        error!(logger, "Error writing to block cache {}", e);
-                    }
                     blocks.extend(new_blocks);
                     blocks.sort_by_key(|block| block.number);
                     stream::iter_ok(blocks)
@@ -2028,6 +2028,9 @@ pub(crate) async fn get_calls(
                 calls: Some(calls),
             }))
         }
+        BlockFinality::Ptr(_) => {
+            unreachable!("get_calls called with BlockFinality::Ptr")
+        }
     }
 }
 
@@ -2209,6 +2212,11 @@ async fn filter_call_triggers_from_unsuccessful_transactions(
                     "this function should not be called when dealing with non-final blocks"
                 )
             }
+            BlockFinality::Ptr(_block) => {
+                unreachable!(
+                    "this function should not be called when dealing with header-only blocks"
+                )
+            }
         }
     };
 

chain/near/src/chain.rs

@@ -325,7 +325,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
         panic!("Should never be called since not used by FirehoseBlockStream")
     }
 
-    async fn load_blocks_by_numbers(
+    async fn load_block_ptrs_by_numbers(
         &self,
         _logger: Logger,
         _block_numbers: HashSet<BlockNumber>,

chain/starknet/src/chain.rs

@@ -375,7 +375,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
         panic!("Should never be called since FirehoseBlockStream cannot resolve it")
     }
 
-    async fn load_blocks_by_numbers(
+    async fn load_block_ptrs_by_numbers(
         &self,
         _logger: Logger,
         _block_numbers: HashSet<BlockNumber>,

chain/substreams/src/trigger.rs

@@ -136,7 +136,7 @@ impl blockchain::TriggersAdapter<Chain> for TriggersAdapter {
         unimplemented!()
     }
 
-    async fn load_blocks_by_numbers(
+    async fn load_block_ptrs_by_numbers(
         &self,
         _logger: Logger,
         _block_numbers: HashSet<BlockNumber>,

graph/src/blockchain/block_stream.rs

@@ -412,7 +412,7 @@ async fn scan_subgraph_triggers<C: Blockchain>(
             block_numbers.insert(to);
 
             let blocks = adapter
-                .load_blocks_by_numbers(logger.clone(), block_numbers)
+                .load_block_ptrs_by_numbers(logger.clone(), block_numbers)
                 .await?;
 
             create_subgraph_triggers::<C>(logger.clone(), blocks, filter, entities).await
@@ -581,7 +581,7 @@ pub trait TriggersAdapter<C: Blockchain>: Send + Sync {
     /// Get pointer to parent of `block`. This is called when reverting `block`.
     async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error>;
 
-    async fn load_blocks_by_numbers(
+    async fn load_block_ptrs_by_numbers(
         &self,
         logger: Logger,
         block_numbers: HashSet<BlockNumber>,

graph/src/blockchain/mock.rs

@@ -233,7 +233,7 @@ impl TriggersAdapter<MockBlockchain> for MockTriggersAdapter {
         todo!()
     }
 
-    async fn load_blocks_by_numbers(
+    async fn load_block_ptrs_by_numbers(
         &self,
         _logger: Logger,
         _block_numbers: HashSet<BlockNumber>,