perf(index-node): batch block hash lookups for POI queries

Pre-fetch all block hashes in a single batch query before parallel POI
processing, reducing database round-trips from 10+ to 1-2 per batch.

- Add block_hashes_by_block_numbers batch method to ChainStore trait
- Add get_public_proof_of_indexing_with_block_hash to StatusStore trait
- Modify resolver to group requests by network and batch-fetch hashes
- Pass pre-fetched hashes to avoid redundant lookups during parallel POI

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: MoonBoi9001

graph/src/components/store/traits.rs

@@ -602,6 +602,12 @@ pub trait ChainStore: ChainHeadStore {
         number: BlockNumber,
     ) -> Result<Vec<BlockHash>, Error>;
 
+    /// Return the hashes of all blocks with the given numbers (batch version)
+    async fn block_hashes_by_block_numbers(
+        &self,
+        numbers: &[BlockNumber],
+    ) -> Result<HashMap<BlockNumber, Vec<BlockHash>>, Error>;
+
     /// Confirm that block number `number` has hash `hash` and that the store
     /// may purge any other blocks with that number
     async fn confirm_block_hash(
@@ -790,6 +796,22 @@ pub trait StatusStore: Send + Sync + 'static {
         block_number: BlockNumber,
         fetch_block_ptr: &dyn BlockPtrForNumber,
     ) -> Result<Option<(PartialBlockPtr, [u8; 32])>, StoreError>;
+
+    /// Like `get_public_proof_of_indexing` but accepts optional pre-fetched block hashes
+    /// to avoid redundant database lookups when processing batches of POI requests.
+    async fn get_public_proof_of_indexing_with_block_hash(
+        &self,
+        subgraph_id: &DeploymentHash,
+        block_number: BlockNumber,
+        prefetched_hashes: Option<&Vec<BlockHash>>,
+        fetch_block_ptr: &dyn BlockPtrForNumber,
+    ) -> Result<Option<(PartialBlockPtr, [u8; 32])>, StoreError>;
+
+    /// Get the network for a deployment
+    async fn network_for_deployment(
+        &self,
+        id: &DeploymentHash,
+    ) -> Result<String, StoreError>;
 }
 
 #[async_trait]

server/index-node/src/resolver.rs

@@ -1,4 +1,4 @@
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashMap};
 
 use async_trait::async_trait;
 use graph::data::query::Trace;
@@ -417,38 +417,101 @@ where
             return Err(QueryExecutionError::TooExpensive);
         }
 
-        // Process all POI requests in parallel for better throughput
-        let poi_futures: Vec<_> = requests
-            .into_iter()
-            .map(|request| async move {
-                let poi_result = match self
-                    .store
-                    .get_public_proof_of_indexing(&request.deployment, request.block_number, self)
+        // Step 1: Group requests by network and collect block numbers for batch lookup
+        let mut requests_by_network: HashMap<String, Vec<(usize, BlockNumber)>> = HashMap::new();
+        let mut request_networks: Vec<Option<String>> = Vec::with_capacity(requests.len());
+
+        for (idx, request) in requests.iter().enumerate() {
+            match self.store.network_for_deployment(&request.deployment).await {
+                Ok(network) => {
+                    requests_by_network
+                        .entry(network.clone())
+                        .or_default()
+                        .push((idx, request.block_number));
+                    request_networks.push(Some(network));
+                }
+                Err(_) => {
+                    request_networks.push(None);
+                }
+            }
+        }
+
+        // Step 2: Pre-fetch all block hashes per network in batch
+        let mut block_hash_cache: HashMap<(String, BlockNumber), Vec<graph::blockchain::BlockHash>> =
+            HashMap::new();
+
+        for (network, network_requests) in &requests_by_network {
+            let block_numbers: Vec<BlockNumber> =
+                network_requests.iter().map(|(_, num)| *num).collect();
+
+            if let Some(chain_store) = self.store.block_store().chain_store(network).await {
+                match chain_store
+                    .block_hashes_by_block_numbers(&block_numbers)
                     .await
                 {
-                    Ok(Some(poi)) => Some(poi),
-                    Ok(None) => None,
+                    Ok(hashes) => {
+                        for (num, hash_vec) in hashes {
+                            block_hash_cache.insert((network.clone(), num), hash_vec);
+                        }
+                    }
                     Err(e) => {
-                        error!(
+                        debug!(
                             self.logger,
-                            "Failed to query public proof of indexing";
-                            "subgraph" => &request.deployment,
-                            "block" => format!("{}", request.block_number),
+                            "Failed to batch fetch block hashes for network";
+                            "network" => network,
                             "error" => format!("{:?}", e)
                         );
-                        None
+                        // Continue without pre-fetched hashes - will fall back to individual lookups
+                    }
+                }
+            }
+        }
+
+        // Step 3: Process all POI requests in parallel, using cached block hashes
+        let poi_futures: Vec<_> = requests
+            .into_iter()
+            .zip(request_networks.into_iter())
+            .map(|(request, network_opt)| {
+                let cache = &block_hash_cache;
+                async move {
+                    let prefetched_hashes = network_opt
+                        .as_ref()
+                        .and_then(|network| cache.get(&(network.clone(), request.block_number)));
+
+                    let poi_result = match self
+                        .store
+                        .get_public_proof_of_indexing_with_block_hash(
+                            &request.deployment,
+                            request.block_number,
+                            prefetched_hashes,
+                            self,
+                        )
+                        .await
+                    {
+                        Ok(Some(poi)) => Some(poi),
+                        Ok(None) => None,
+                        Err(e) => {
+                            error!(
+                                self.logger,
+                                "Failed to query public proof of indexing";
+                                "subgraph" => &request.deployment,
+                                "block" => format!("{}", request.block_number),
+                                "error" => format!("{:?}", e)
+                            );
+                            None
+                        }
+                    };
+
+                    PublicProofOfIndexingResult {
+                        deployment: request.deployment,
+                        block: match poi_result {
+                            Some((ref block, _)) => block.clone(),
+                            None => PartialBlockPtr::from(request.block_number),
+                        },
+                        proof_of_indexing: poi_result.map(|(_, poi)| poi),
                     }
-                };
-
-                PublicProofOfIndexingResult {
-                    deployment: request.deployment,
-                    block: match poi_result {
-                        Some((ref block, _)) => block.clone(),
-                        None => PartialBlockPtr::from(request.block_number),
-                    },
-                    proof_of_indexing: poi_result.map(|(_, poi)| poi),
+                    .into_value()
                 }
-                .into_value()
             })
             .collect();
 

store/postgres/src/chain_store.rs

@@ -739,6 +739,60 @@ mod data {
             }
         }
 
+        /// Return the hashes of all blocks with the given block numbers (batch version)
+        pub(super) async fn block_hashes_by_block_numbers(
+            &self,
+            conn: &mut AsyncPgConnection,
+            chain: &str,
+            numbers: &[BlockNumber],
+        ) -> Result<HashMap<BlockNumber, Vec<BlockHash>>, Error> {
+            if numbers.is_empty() {
+                return Ok(HashMap::new());
+            }
+
+            match self {
+                Storage::Shared => {
+                    use public::ethereum_blocks as b;
+
+                    let results = b::table
+                        .select((b::number, b::hash))
+                        .filter(b::network_name.eq(chain))
+                        .filter(
+                            b::number.eq_any(Vec::from_iter(numbers.iter().map(|&n| n as i64))),
+                        )
+                        .load::<(i64, String)>(conn)
+                        .await?;
+
+                    let mut map: HashMap<BlockNumber, Vec<BlockHash>> = HashMap::new();
+                    for (num, hash) in results {
+                        let block_hash = hash.parse()?;
+                        map.entry(num as BlockNumber).or_default().push(block_hash);
+                    }
+                    Ok(map)
+                }
+                Storage::Private(Schema { blocks, .. }) => {
+                    let results = blocks
+                        .table()
+                        .select((blocks.number(), blocks.hash()))
+                        .filter(
+                            blocks
+                                .number()
+                                .eq_any(Vec::from_iter(numbers.iter().map(|&n| n as i64))),
+                        )
+                        .load::<(i64, Vec<u8>)>(conn)
+                        .await?;
+
+                    let mut map: HashMap<BlockNumber, Vec<BlockHash>> = HashMap::new();
+                    for (num, hash) in results {
+                        map.entry(num as BlockNumber)
+                            .or_default()
+                            .push(BlockHash::from(hash));
+                    }
+                    Ok(map)
+                }
+            }
+        }
+
         pub(super) async fn confirm_block_hash(
             &self,
             conn: &mut AsyncPgConnection,
@@ -2971,6 +3025,16 @@ impl ChainStoreTrait for ChainStore {
             .await
     }
 
+    async fn block_hashes_by_block_numbers(
+        &self,
+        numbers: &[BlockNumber],
+    ) -> Result<HashMap<BlockNumber, Vec<BlockHash>>, Error> {
+        let mut conn = self.pool.get_permitted().await?;
+        self.storage
+            .block_hashes_by_block_numbers(&mut conn, &self.chain, numbers)
+            .await
+    }
+
     async fn confirm_block_hash(
         &self,
         number: BlockNumber,

store/postgres/src/store.rs

@@ -171,6 +171,31 @@ impl StatusStore for Store {
             .await
     }
 
+    async fn get_public_proof_of_indexing_with_block_hash(
+        &self,
+        subgraph_id: &DeploymentHash,
+        block_number: BlockNumber,
+        prefetched_hashes: Option<&Vec<graph::blockchain::BlockHash>>,
+        fetch_block_ptr: &dyn BlockPtrForNumber,
+    ) -> Result<Option<(PartialBlockPtr, [u8; 32])>, StoreError> {
+        self.subgraph_store
+            .get_public_proof_of_indexing_with_block_hash(
+                subgraph_id,
+                block_number,
+                prefetched_hashes,
+                self.block_store().clone(),
+                fetch_block_ptr,
+            )
+            .await
+    }
+
+    async fn network_for_deployment(
+        &self,
+        id: &DeploymentHash,
+    ) -> Result<String, StoreError> {
+        self.subgraph_store.network_for_deployment(id).await
+    }
+
     async fn query_permit(&self) -> QueryPermit {
         // Status queries go to the primary shard.
         self.block_store.query_permit_primary().await