Subgraph Composition: Option to force rpc to fetch block ptrs (#5876)

* chain/ethereum:  Add parallel block fetching with configurable batch size when using firehose with composable subgraphs

* ethereum: Add option to force RPC for block pointer lookups

This adds GRAPH_ETHEREUM_FORCE_RPC_FOR_BLOCK_PTRS env var which when enabled forces the use of RPC instead of Firehose for loading block pointers by numbers, with Firehose fallback. Useful for composable subgraphs.

* graph: change log level for get_block_by_number

* graph: Add get_block_number_with_retry method for firehose endpoint

* Address review comments

Author: incrypto32

chain/ethereum/src/chain.rs

@@ -17,7 +17,7 @@ use graph::prelude::{
     EthereumCallCache, LightEthereumBlock, LightEthereumBlockExt, MetricsRegistry,
 };
 use graph::schema::InputSchema;
-use graph::slog::{debug, error, trace};
+use graph::slog::{debug, error, trace, warn};
 use graph::substreams::Clock;
 use graph::{
     blockchain::{
@@ -257,6 +257,7 @@ pub struct EthereumAdapterSelector {
     client: Arc<ChainClient<Chain>>,
     registry: Arc<MetricsRegistry>,
     chain_store: Arc<dyn ChainStore>,
+    eth_adapters: Arc<EthereumNetworkAdapters>,
 }
 
 impl EthereumAdapterSelector {
@@ -265,12 +266,14 @@ impl EthereumAdapterSelector {
         client: Arc<ChainClient<Chain>>,
         registry: Arc<MetricsRegistry>,
         chain_store: Arc<dyn ChainStore>,
+        eth_adapters: Arc<EthereumNetworkAdapters>,
     ) -> Self {
         Self {
             logger_factory,
             client,
             registry,
             chain_store,
+            eth_adapters,
         }
     }
 }
@@ -296,6 +299,7 @@ impl TriggersAdapterSelector<Chain> for EthereumAdapterSelector {
             chain_store: self.chain_store.cheap_clone(),
             unified_api_version,
             capabilities: *capabilities,
+            eth_adapters: self.eth_adapters.cheap_clone(),
         };
         Ok(Arc::new(adapter))
     }
@@ -739,6 +743,7 @@ pub struct TriggersAdapter {
     chain_client: Arc<ChainClient<Chain>>,
     capabilities: NodeCapabilities,
     unified_api_version: UnifiedMappingApiVersion,
+    eth_adapters: Arc<EthereumNetworkAdapters>,
 }
 
 /// Fetches blocks from the cache based on block numbers, excluding duplicates
@@ -784,12 +789,34 @@ async fn fetch_unique_blocks_from_cache(
             "Loading {} block(s) not in the block cache",
             missing_blocks.len()
         );
-        debug!(logger, "Missing blocks {:?}", missing_blocks);
+        trace!(logger, "Missing blocks {:?}", missing_blocks.len());
     }
 
     (blocks, missing_blocks)
 }
 
+// This is used to load blocks from the RPC.
+async fn load_blocks_with_rpc(
+    logger: &Logger,
+    adapter: Arc<EthereumAdapter>,
+    chain_store: Arc<dyn ChainStore>,
+    block_numbers: BTreeSet<BlockNumber>,
+) -> Result<Vec<BlockFinality>> {
+    let logger_clone = logger.clone();
+    load_blocks(
+        logger,
+        chain_store,
+        block_numbers,
+        |missing_numbers| async move {
+            adapter
+                .load_block_ptrs_by_numbers_rpc(logger_clone, missing_numbers)
+                .try_collect()
+                .await
+        },
+    )
+    .await
+}
+
 /// Fetches blocks by their numbers, first attempting to load from cache.
 /// Missing blocks are retrieved from an external source, with all blocks sorted and converted to `BlockFinality` format.
 async fn load_blocks<F, Fut>(
@@ -847,6 +874,37 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
     ) -> Result<Vec<BlockFinality>> {
         match &*self.chain_client {
             ChainClient::Firehose(endpoints) => {
+                // If the force_rpc_for_block_ptrs flag is set, we will use the RPC to load the blocks
+                // even if the firehose is available. If no adapter is available, we will log an error.
+                // And then fallback to the firehose.
+                if ENV_VARS.force_rpc_for_block_ptrs {
+                    trace!(
+                        logger,
+                        "Loading blocks from RPC (force_rpc_for_block_ptrs is set)";
+                        "block_numbers" => format!("{:?}", block_numbers)
+                    );
+                    match self.eth_adapters.cheapest_with(&self.capabilities).await {
+                        Ok(adapter) => {
+                            match load_blocks_with_rpc(
+                                &logger,
+                                adapter,
+                                self.chain_store.clone(),
+                                block_numbers.clone(),
+                            )
+                            .await
+                            {
+                                Ok(blocks) => return Ok(blocks),
+                                Err(e) => {
+                                    warn!(logger, "Error loading blocks from RPC: {}", e);
+                                }
+                            }
+                        }
+                        Err(e) => {
+                            warn!(logger, "Error getting cheapest adapter: {}", e);
+                        }
+                    }
+                }
+
                 trace!(
                     logger,
                     "Loading blocks from firehose";
@@ -884,29 +942,16 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
                 .await
             }
 
-            ChainClient::Rpc(client) => {
+            ChainClient::Rpc(eth_adapters) => {
                 trace!(
                     logger,
                     "Loading blocks from RPC";
                     "block_numbers" => format!("{:?}", block_numbers)
                 );
 
-                let adapter = client.cheapest_with(&self.capabilities).await?;
-                let chain_store = self.chain_store.clone();
-                let logger_clone = logger.clone();
-
-                load_blocks(
-                    &logger,
-                    chain_store,
-                    block_numbers,
-                    |missing_numbers| async move {
-                        adapter
-                            .load_block_ptrs_by_numbers_rpc(logger_clone, missing_numbers)
-                            .try_collect()
-                            .await
-                    },
-                )
-                .await
+                let adapter = eth_adapters.cheapest_with(&self.capabilities).await?;
+                load_blocks_with_rpc(&logger, adapter, self.chain_store.clone(), block_numbers)
+                    .await
             }
         }
     }
@@ -973,10 +1018,12 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
             ChainClient::Firehose(endpoints) => {
                 let endpoint = endpoints.endpoint().await?;
                 let block = endpoint
-                    .get_block_by_number::<codec::Block>(ptr.number as u64, &self.logger)
+                    .get_block_by_number_with_retry::<codec::Block>(ptr.number as u64, &self.logger)
                     .await
-                    .map_err(|e| anyhow!("Failed to fetch block from firehose: {}", e))?;
-
+                    .context(format!(
+                        "Failed to fetch block {} from firehose",
+                        ptr.number
+                    ))?;
                 Ok(block.hash() == ptr.hash)
             }
             ChainClient::Rpc(adapter) => {

chain/ethereum/src/env.rs

@@ -91,6 +91,10 @@ pub struct EnvVars {
     /// This is a comma separated list of chain ids for which the gas field will not be set
     /// when calling `eth_call`.
     pub eth_call_no_gas: Vec<String>,
+    /// Set by the flag `GRAPH_ETHEREUM_FORCE_RPC_FOR_BLOCK_PTRS`. On by default.
+    /// When enabled, forces the use of RPC instead of Firehose for loading block pointers by numbers.
+    /// This is used in composable subgraphs. Firehose can be slow for loading block pointers by numbers.
+    pub force_rpc_for_block_ptrs: bool,
 }
 
 // This does not print any values avoid accidentally leaking any sensitive env vars
@@ -141,6 +145,7 @@ impl From<Inner> for EnvVars {
                 .filter(|s| !s.is_empty())
                 .map(str::to_string)
                 .collect(),
+            force_rpc_for_block_ptrs: x.force_rpc_for_block_ptrs.0,
         }
     }
 }
@@ -192,4 +197,6 @@ struct Inner {
     genesis_block_number: u64,
     #[envconfig(from = "GRAPH_ETH_CALL_NO_GAS", default = "421613,421614")]
     eth_call_no_gas: String,
+    #[envconfig(from = "GRAPH_ETHEREUM_FORCE_RPC_FOR_BLOCK_PTRS", default = "true")]
+    force_rpc_for_block_ptrs: EnvVarBoolean,
 }

graph/src/env/mod.rs

@@ -247,6 +247,9 @@ pub struct EnvVars {
     /// Set by the environment variable `GRAPH_FIREHOSE_FETCH_BLOCK_TIMEOUT_SECS`.
     /// The default value is 60 seconds.
     pub firehose_block_fetch_timeout: u64,
+    /// Set by the environment variable `GRAPH_FIREHOSE_BLOCK_BATCH_SIZE`.
+    /// The default value is 10.
+    pub firehose_block_batch_size: usize,
 }
 
 impl EnvVars {
@@ -339,6 +342,7 @@ impl EnvVars {
             block_write_capacity: inner.block_write_capacity.0,
             firehose_block_fetch_retry_limit: inner.firehose_block_fetch_retry_limit,
             firehose_block_fetch_timeout: inner.firehose_block_fetch_timeout,
+            firehose_block_batch_size: inner.firehose_block_fetch_batch_size,
         })
     }
 
@@ -506,6 +510,8 @@ struct Inner {
     firehose_block_fetch_retry_limit: usize,
     #[envconfig(from = "GRAPH_FIREHOSE_FETCH_BLOCK_TIMEOUT_SECS", default = "60")]
     firehose_block_fetch_timeout: u64,
+    #[envconfig(from = "GRAPH_FIREHOSE_FETCH_BLOCK_BATCH_SIZE", default = "10")]
+    firehose_block_fetch_batch_size: usize,
 }
 
 #[derive(Clone, Debug)]