Subgraph composition: Use block ptrs instead of full blocks in blockstream

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,6 +1,7 @@
 use anyhow::Error;
 use ethabi::{Error as ABIError, Function, ParamType, Token};
 use graph::blockchain::ChainIdentifier;
+use graph::blockchain::ExtendedBlockPtr;
 use graph::components::subgraph::MappingError;
 use graph::data::store::ethereum::call;
 use graph::firehose::CallToFilter;
@@ -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<ExtendedBlockPtr>, 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,8 +3,8 @@ 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,
-    TriggersAdapterSelector,
+    BlockIngestor, BlockTime, BlockchainKind, ChainIdentifier, ExtendedBlockPtr,
+    TriggerFilterWrapper, TriggersAdapterSelector,
 };
 use graph::components::network_provider::ChainName;
 use graph::components::store::{DeploymentCursorTracker, SourceableStore};
@@ -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,32 @@ 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"),
+            _ if is_using_subgraph_composition => chain.reorg_threshold,
+            _ => panic!(
+                "expected rpc when using polling blockstream : {}",
+                is_using_subgraph_composition
+            ),
         };
-        let reorg_threshold = match CELO_CHAIN_IDS.contains(&chain_id) {
-            false => chain.reorg_threshold,
-            true => 0,
+
+        let max_block_range_size = if is_using_subgraph_composition {
+            ENV_VARS.max_block_range_size * 10
+        } else {
+            ENV_VARS.max_block_range_size
         };
 
         Ok(Box::new(PollingBlockStream::new(
@@ -207,7 +220,7 @@ impl BlockStreamBuilder<Chain> for EthereumStreamBuilder {
             start_blocks,
             reorg_threshold,
             logger,
-            ENV_VARS.max_block_range_size,
+            max_block_range_size,
             ENV_VARS.target_triggers_per_block_range,
             unified_api_version,
             subgraph_current_block,
@@ -617,6 +630,8 @@ pub enum BlockFinality {
 
     // If a block may still be reorged, we need to work with more local data.
     NonFinal(EthereumBlockWithCalls),
+
+    Ptr(Arc<ExtendedBlockPtr>),
 }
 
 impl Default for BlockFinality {
@@ -630,6 +645,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 +655,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 +665,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 +708,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 +758,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 +772,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 +835,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/env.rs

@@ -33,6 +33,9 @@ pub struct EnvVars {
     /// Set by the environment variable `ETHEREUM_BLOCK_BATCH_SIZE`. The
     /// default value is 10 blocks.
     pub block_batch_size: usize,
+    /// Set by the environment variable `ETHEREUM_BLOCK_PTR_BATCH_SIZE`. The
+    /// default value is 10 blocks.
+    pub block_ptr_batch_size: usize,
     /// Maximum number of blocks to request in each chunk.
     ///
     /// Set by the environment variable `GRAPH_ETHEREUM_MAX_BLOCK_RANGE_SIZE`.
@@ -116,6 +119,7 @@ impl From<Inner> for EnvVars {
             trace_stream_step_size: x.trace_stream_step_size,
             max_event_only_range: x.max_event_only_range,
             block_batch_size: x.block_batch_size,
+            block_ptr_batch_size: x.block_ptr_batch_size,
             max_block_range_size: x.max_block_range_size,
             json_rpc_timeout: Duration::from_secs(x.json_rpc_timeout_in_secs),
             block_receipts_check_timeout: Duration::from_secs(
@@ -160,6 +164,8 @@ struct Inner {
     max_event_only_range: BlockNumber,
     #[envconfig(from = "ETHEREUM_BLOCK_BATCH_SIZE", default = "10")]
     block_batch_size: usize,
+    #[envconfig(from = "ETHEREUM_BLOCK_PTR_BATCH_SIZE", default = "100")]
+    block_ptr_batch_size: usize,
     #[envconfig(from = "GRAPH_ETHEREUM_MAX_BLOCK_RANGE_SIZE", default = "2000")]
     max_block_range_size: BlockNumber,
     #[envconfig(from = "GRAPH_ETHEREUM_JSON_RPC_TIMEOUT", default = "180")]

chain/ethereum/src/ethereum_adapter.rs

@@ -2,6 +2,7 @@ use futures03::{future::BoxFuture, stream::FuturesUnordered};
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::BlockHash;
 use graph::blockchain::ChainIdentifier;
+use graph::blockchain::ExtendedBlockPtr;
 
 use graph::components::transaction_receipt::LightTransactionReceipt;
 use graph::data::store::ethereum::call;
@@ -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<ExtendedBlockPtr>, Error = Error> + Send {
         let web3 = self.web3.clone();
 
         stream::iter_ok::<_, Error>(numbers.into_iter().map(move |number| {
@@ -798,27 +799,37 @@ 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 = ExtendedBlockPtr::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()
                 })
                 .boxed()
                 .compat()
                 .from_err()
         }))
-        .buffered(ENV_VARS.block_batch_size)
+        .buffered(ENV_VARS.block_ptr_batch_size)
     }
 
     /// Request blocks ptrs for numbers through JSON-RPC.
@@ -1690,27 +1701,27 @@ 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> {
-        let blocks_map: BTreeMap<i32, Vec<json::Value>> = chain_store
+    ) -> Box<dyn Stream<Item = Arc<ExtendedBlockPtr>, Error = Error> + Send> {
+        let blocks_map = 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<ExtendedBlockPtr>> = blocks_map
             .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
                 }
@@ -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>,