feat(index-node): add blockForPoi query for dispute investigation

MoonBoi9001 · claude · MoonBoi9001 · commit 481fbb523f58 · 2026-02-05T22:39:05.000-04:00
Adds a new index-node GraphQL query that searches a block range to find
which block produced a given POI. This supports dispute investigation
where an indexer may have submitted a POI for block N that actually
corresponds to block X &lt; N.

The resolver fetches all poi2$ digest entries in a single DB call, batch
fetches block hashes in 50k chunks, and runs ProofOfIndexingFinisher for
each block until a match is found.

Also adds network_for_deployment to the StatusStore trait and its
implementation chain (SubgraphStore, Store) as a supporting method
for resolving the chain store from a deployment hash.

Co-Authored-By: Claude Opus 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/graph/src/components/store/traits.rs b/graph/src/components/store/traits.rs
@@ -10,6 +10,7 @@ use crate::blockchain::{BlockTime, ChainIdentifier, ExtendedBlockPtr};
 use crate::components::metrics::stopwatch::StopwatchMetrics;
 use crate::components::network_provider::ChainName;
 use crate::components::server::index_node::VersionInfo;
+use crate::components::subgraph::ProofOfIndexingVersion;
 use crate::components::subgraph::SubgraphVersionSwitchingMode;
 use crate::components::transaction_receipt;
 use crate::components::versions::ApiVersion;
@@ -741,6 +742,28 @@ pub trait QueryStore: Send + Sync {
     fn deployment_id(&self) -> DeploymentId;
 }
 
+/// A single POI digest entry from the `poi2$` table, representing the
+/// accumulated digest for a causality region over a block range.
+#[derive(Clone, Debug)]
+pub struct PoiDigestEntry {
+    /// The causality region identifier (the entity id in poi2$)
+    pub id: Id,
+    /// The accumulated digest bytes
+    pub digest: Vec<u8>,
+    /// Start of the block range (inclusive)
+    pub start_block: BlockNumber,
+    /// End of the block range (exclusive, i32::MAX if open-ended)
+    pub end_block: BlockNumber,
+}
+
+/// The full POI digest history for a deployment, containing all digest
+/// entries and the POI version needed to compute proofs.
+#[derive(Clone, Debug)]
+pub struct PoiDigestHistory {
+    pub entries: Vec<PoiDigestEntry>,
+    pub poi_version: ProofOfIndexingVersion,
+}
+
 /// A view of the store that can provide information about the indexing status
 /// of any subgraph and any deployment
 #[async_trait]
@@ -790,6 +813,19 @@ pub trait StatusStore: Send + Sync + 'static {
         block_number: BlockNumber,
         fetch_block_ptr: &dyn BlockPtrForNumber,
     ) -> Result<Option<(PartialBlockPtr, [u8; 32])>, StoreError>;
+
+    /// Retrieve the full POI digest history for a deployment within a block
+    /// range. Returns all `poi2$` entries whose block ranges overlap the
+    /// given range, along with the deployment's `ProofOfIndexingVersion`.
+    /// Returns `None` if the deployment doesn't exist or has no POI data.
+    async fn get_poi_digest_history(
+        &self,
+        subgraph_id: &DeploymentHash,
+        block_range: std::ops::Range<BlockNumber>,
+    ) -> Result<Option<PoiDigestHistory>, StoreError>;
+
+    /// Get the network for a deployment
+    async fn network_for_deployment(&self, id: &DeploymentHash) -> Result<String, StoreError>;
 }
 
 #[async_trait]
diff --git a/server/index-node/src/resolver.rs b/server/index-node/src/resolver.rs
@@ -1,6 +1,7 @@
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashMap};
 
 use async_trait::async_trait;
+use graph::components::subgraph::ProofOfIndexingFinisher;
 use graph::data::query::Trace;
 use graph::data::store::Id;
 use graph::prelude::alloy::primitives::Address;
@@ -454,6 +455,183 @@ where
         Ok(r::Value::List(public_poi_results))
     }
 
+    async fn resolve_block_for_poi(
+        &self,
+        field: &a::Field,
+    ) -> Result<r::Value, QueryExecutionError> {
+        const MAX_BLOCK_RANGE: i32 = 1_000_000;
+        const BLOCK_HASH_CHUNK_SIZE: usize = 50_000;
+
+        let deployment_id = field
+            .get_required::<DeploymentHash>("subgraph")
+            .expect("Valid subgraph required");
+        let target_poi_hash = field
+            .get_required::<BlockHash>("targetPoi")
+            .expect("Valid targetPoi required");
+        let start_block = field
+            .get_required::<BlockNumber>("startBlock")
+            .expect("Valid startBlock required");
+        let end_block = field
+            .get_required::<BlockNumber>("endBlock")
+            .expect("Valid endBlock required");
+
+        if end_block <= start_block {
+            return Ok(r::Value::Null);
+        }
+
+        if end_block - start_block > MAX_BLOCK_RANGE {
+            return Err(QueryExecutionError::TooExpensive);
+        }
+
+        let target_bytes: [u8; 32] = match target_poi_hash.as_slice().try_into() {
+            Ok(bytes) => bytes,
+            Err(_) => {
+                error!(
+                    self.logger,
+                    "Invalid targetPoi: expected 32 bytes";
+                    "got_bytes" => target_poi_hash.as_slice().len()
+                );
+                return Ok(r::Value::Null);
+            }
+        };
+
+        // Resolve the network for this deployment
+        let network = match self.store.network_for_deployment(&deployment_id).await {
+            Ok(n) => n,
+            Err(e) => {
+                error!(
+                    self.logger,
+                    "Failed to resolve network for deployment";
+                    "subgraph" => &deployment_id,
+                    "error" => format!("{:?}", e)
+                );
+                return Ok(r::Value::Null);
+            }
+        };
+
+        // Fetch the full digest history for the block range
+        let history = match self
+            .store
+            .get_poi_digest_history(&deployment_id, start_block..end_block)
+            .await
+        {
+            Ok(Some(h)) => h,
+            Ok(None) => return Ok(r::Value::Null),
+            Err(e) => {
+                error!(
+                    self.logger,
+                    "Failed to fetch POI digest history";
+                    "subgraph" => &deployment_id,
+                    "error" => format!("{:?}", e)
+                );
+                return Ok(r::Value::Null);
+            }
+        };
+
+        let poi_version = history.poi_version;
+
+        // Build a lookup structure: for each causality region id, a sorted
+        // vec of (start_block, end_block, digest) for binary search.
+        let mut region_entries: HashMap<Id, Vec<(BlockNumber, BlockNumber, Vec<u8>)>> =
+            HashMap::new();
+        for entry in history.entries {
+            region_entries.entry(entry.id).or_default().push((
+                entry.start_block,
+                entry.end_block,
+                entry.digest,
+            ));
+        }
+        // Entries are already sorted by (id, start_block) from the SQL query,
+        // but sort each region's vec to be safe.
+        for entries in region_entries.values_mut() {
+            entries.sort_by_key(|(start, _, _)| *start);
+        }
+
+        let chain_store = match self.store.block_store().chain_store(&network).await {
+            Some(cs) => cs,
+            None => {
+                error!(
+                    self.logger,
+                    "Chain store not found for network";
+                    "network" => &network
+                );
+                return Ok(r::Value::Null);
+            }
+        };
+
+        // Process blocks in chunks to avoid excessive memory usage
+        let mut current = start_block;
+        while current < end_block {
+            let chunk_end = std::cmp::min(current + BLOCK_HASH_CHUNK_SIZE as i32, end_block);
+
+            // Batch-fetch block hashes for this chunk using block_ptrs_by_numbers
+            let block_numbers: Vec<BlockNumber> = (current..chunk_end).collect();
+            let block_ptrs = match chain_store
+                .cheap_clone()
+                .block_ptrs_by_numbers(block_numbers)
+                .await
+            {
+                Ok(ptrs) => ptrs,
+                Err(e) => {
+                    error!(
+                        self.logger,
+                        "Failed to fetch block hashes";
+                        "network" => &network,
+                        "range" => format!("{}..{}", current, chunk_end),
+                        "error" => format!("{:?}", e)
+                    );
+                    return Ok(r::Value::Null);
+                }
+            };
+
+            for block_num in current..chunk_end {
+                let ptrs = match block_ptrs.get(&block_num) {
+                    Some(p) if p.len() == 1 => p,
+                    _ => continue, // Skip blocks with no hash or ambiguous hashes
+                };
+                let block_hash = &ptrs[0].hash;
+                let block_ptr = BlockPtr::new(block_hash.clone(), block_num);
+
+                // Compute POI for this block using digest entries
+                let mut finisher = ProofOfIndexingFinisher::new(
+                    &block_ptr,
+                    &deployment_id,
+                    &Some(Address::ZERO),
+                    poi_version,
+                );
+
+                for (region_id, entries) in &region_entries {
+                    // Binary search for the entry that covers this block number
+                    let idx = entries.partition_point(|(start, _, _)| *start <= block_num);
+                    if idx == 0 {
+                        continue;
+                    }
+                    let (start, end, ref digest) = entries[idx - 1];
+                    if block_num >= start && block_num < end {
+                        finisher.add_causality_region(region_id, digest);
+                    }
+                }
+
+                let computed_poi = finisher.finish();
+                if computed_poi == target_bytes {
+                    return Ok(object! {
+                        __typename: "PoiSearchResult",
+                        deployment: deployment_id.to_string(),
+                        block: object! {
+                            hash: block_hash.hash_hex(),
+                            number: block_num,
+                        },
+                        proofOfIndexing: format!("0x{}", hex::encode(computed_poi)),
+                    });
+                }
+            }
+
+            current = chunk_end;
+        }
+
+        Ok(r::Value::Null)
+    }
+
     async fn resolve_indexing_status_for_version(
         &self,
         field: &a::Field,
@@ -858,6 +1036,7 @@ where
             // The top-level `subgraphVersions` field
             (None, "apiVersions") => self.resolve_api_versions(field),
             (None, "version") => self.version(),
+            (None, "blockForPoi") => self.resolve_block_for_poi(field).await,
 
             // Resolve fields of `Object` values (e.g. the `latestBlock` field of `EthereumBlock`)
             (value, _) => Ok(value.unwrap_or(r::Value::Null)),
diff --git a/server/index-node/src/schema.graphql b/server/index-node/src/schema.graphql
@@ -46,6 +46,17 @@ type Query {
     blockHash: Bytes!
   ): [CachedEthereumCall!]
   apiVersions(subgraphId: String!): [ApiVersion!]!
+  """
+  Find the block number that produced a given proof of indexing.
+  Used for dispute investigation to verify which block an indexer
+  actually synced to when they submitted a POI.
+  """
+  blockForPoi(
+    subgraph: String!
+    targetPoi: Bytes!
+    startBlock: Int!
+    endBlock: Int!
+  ): PoiSearchResult
 }
 
 type Version {
@@ -203,6 +214,12 @@ type ProofOfIndexingResult {
   proofOfIndexing: Bytes
 }
 
+type PoiSearchResult {
+  block: Block!
+  deployment: String!
+  proofOfIndexing: Bytes!
+}
+
 type ApiVersion {
   """
   Version number in SemVer format
diff --git a/store/postgres/src/deployment_store.rs b/store/postgres/src/deployment_store.rs
@@ -8,14 +8,15 @@ use graph::anyhow::Context;
 use graph::blockchain::block_stream::{EntitySourceOperation, FirehoseCursor};
 use graph::blockchain::BlockTime;
 use graph::components::store::write::RowGroup;
+use graph::components::store::PoiDigestHistory;
 use graph::components::store::{
     Batch, DeploymentLocator, DerivedEntityQuery, PrunePhase, PruneReporter, PruneRequest,
     PruningStrategy, QueryPermit, StoredDynamicDataSource, VersionStats,
 };
 use graph::components::versions::VERSIONS;
 use graph::data::graphql::IntoValue;
 use graph::data::query::Trace;
-use graph::data::store::{IdList, SqlQueryObject};
+use graph::data::store::{Id, IdList, SqlQueryObject};
 use graph::data::subgraph::{status, SPEC_VERSION_0_0_6};
 use graph::data_source::CausalityRegion;
 use graph::derive::CheapClone;
@@ -1032,6 +1033,69 @@ impl DeploymentStore {
         Ok(Some(finisher.finish()))
     }
 
+    /// Retrieve all POI digest entries from the `poi2$` table whose block
+    /// ranges overlap the given `block_range`, along with the deployment's
+    /// POI version. Used by the `blockForPoi` resolver to reconstruct POIs
+    /// without per-block entity queries.
+    pub(crate) async fn get_poi_digest_history(
+        &self,
+        site: Arc<Site>,
+        block_range: Range<BlockNumber>,
+    ) -> Result<Option<PoiDigestHistory>, StoreError> {
+        use diesel::sql_types::{Binary, Integer, Text};
+        use graph::components::store::PoiDigestEntry;
+
+        let info = self.subgraph_info(site.cheap_clone()).await?;
+
+        #[derive(QueryableByName)]
+        struct DigestRow {
+            #[diesel(sql_type = Text)]
+            id: String,
+            #[diesel(sql_type = Binary)]
+            digest: Vec<u8>,
+            #[diesel(sql_type = Integer)]
+            start_block: i32,
+            #[diesel(sql_type = Integer)]
+            end_block: i32,
+        }
+
+        let query = format!(
+            r#"SELECT id, digest, lower(block_range) as start_block,
+                      coalesce(upper(block_range), 2147483647) as end_block
+               FROM "{}"."poi2$"
+               WHERE block_range && int4range($1, $2)
+               ORDER BY id, lower(block_range)"#,
+            site.namespace,
+        );
+
+        let mut conn = self.pool.get_permitted().await?;
+        let rows = diesel::sql_query(query)
+            .bind::<Integer, _>(block_range.start)
+            .bind::<Integer, _>(block_range.end)
+            .load::<DigestRow>(&mut conn)
+            .await
+            .map_err(StoreError::from)?;
+
+        if rows.is_empty() {
+            return Ok(None);
+        }
+
+        let entries = rows
+            .into_iter()
+            .map(|row| PoiDigestEntry {
+                id: Id::String(row.id.into()),
+                digest: row.digest,
+                start_block: row.start_block,
+                end_block: row.end_block,
+            })
+            .collect();
+
+        Ok(Some(PoiDigestHistory {
+            entries,
+            poi_version: info.poi_version,
+        }))
+    }
+
     /// Get the entity matching `key` from the deployment `site`. Only
     /// consider entities as of the given `block`
     pub(crate) async fn get(
diff --git a/store/postgres/src/store.rs b/store/postgres/src/store.rs
@@ -171,6 +171,20 @@ impl StatusStore for Store {
             .await
     }
 
+    async fn get_poi_digest_history(
+        &self,
+        subgraph_id: &DeploymentHash,
+        block_range: std::ops::Range<BlockNumber>,
+    ) -> Result<Option<graph::components::store::PoiDigestHistory>, StoreError> {
+        self.subgraph_store
+            .get_poi_digest_history(subgraph_id, block_range)
+            .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
diff --git a/store/postgres/src/subgraph_store.rs b/store/postgres/src/subgraph_store.rs
