fix false positive POI (#3951)

Author: mangas

core/src/subgraph/trigger_processor.rs

@@ -3,7 +3,7 @@ use graph::blockchain::Blockchain;
 use graph::cheap_clone::CheapClone;
 use graph::components::store::SubgraphFork;
 use graph::components::subgraph::{MappingError, SharedProofOfIndexing};
-use graph::data_source::TriggerData;
+use graph::data_source::{MappingTrigger, TriggerData, TriggerWithHandler};
 use graph::prelude::tokio::time::Instant;
 use graph::prelude::{
     BlockState, RuntimeHost, RuntimeHostBuilder, SubgraphInstanceMetrics, TriggerProcessor,
@@ -33,11 +33,7 @@ where
     ) -> Result<BlockState<C>, MappingError> {
         let error_count = state.deterministic_errors.len();
 
-        if let Some(proof_of_indexing) = proof_of_indexing {
-            proof_of_indexing
-                .borrow_mut()
-                .start_handler(causality_region);
-        }
+        let mut host_mapping: Vec<(&T::Host, TriggerWithHandler<MappingTrigger<C>>)> = vec![];
 
         for host in hosts {
             let mapping_trigger = match host.match_and_decode(trigger, block, logger)? {
@@ -48,6 +44,20 @@ where
                 None => continue,
             };
 
+            host_mapping.push((&host, mapping_trigger));
+        }
+
+        if host_mapping.is_empty() {
+            return Ok(state);
+        }
+
+        if let Some(proof_of_indexing) = proof_of_indexing {
+            proof_of_indexing
+                .borrow_mut()
+                .start_handler(causality_region);
+        }
+
+        for (host, mapping_trigger) in host_mapping {
             let start = Instant::now();
             state = host
                 .process_mapping_trigger(

store/postgres/src/subgraph_store.rs

@@ -228,7 +228,7 @@ impl SubgraphStore {
         }
     }
 
-    pub(crate) async fn get_proof_of_indexing(
+    pub async fn get_proof_of_indexing(
         &self,
         id: &DeploymentHash,
         indexer: &Option<Address>,

tests/integration-tests/dynamic-data-source/abis/Contract.abi

@@ -0,0 +1,33 @@
+[
+  {
+    "inputs": [],
+    "stateMutability": "nonpayable",
+    "type": "constructor"
+  },
+  {
+    "anonymous": false,
+    "inputs": [
+      {
+        "indexed": false,
+        "internalType": "uint16",
+        "name": "x",
+        "type": "uint16"
+      }
+    ],
+    "name": "Trigger",
+    "type": "event"
+  },
+  {
+    "inputs": [
+      {
+        "internalType": "uint16",
+        "name": "x",
+        "type": "uint16"
+      }
+    ],
+    "name": "emitTrigger",
+    "outputs": [],
+    "stateMutability": "nonpayable",
+    "type": "function"
+  }
+]

tests/integration-tests/dynamic-data-source/package.json

@@ -0,0 +1,25 @@
+{
+  "name": "dynamic-data-source",
+  "version": "0.1.0",
+  "scripts": {
+    "build-contracts": "../common/build-contracts.sh",
+    "codegen": "graph codegen",
+    "test": "yarn build-contracts && truffle test --compile-none --network test",
+    "create:test": "graph create test/dynamic-data-source --node $GRAPH_NODE_ADMIN_URI",
+    "deploy:test": "graph deploy test/dynamic-data-source --version-label v0.0.1 --ipfs $IPFS_URI --node $GRAPH_NODE_ADMIN_URI"
+  },
+  "devDependencies": {
+    "@graphprotocol/graph-cli": "https://github.com/graphprotocol/graph-cli#main",
+    "@graphprotocol/graph-ts": "https://github.com/graphprotocol/graph-ts#main",
+    "solc": "^0.8.2"
+  },
+  "dependencies": {
+    "@truffle/contract": "^4.3",
+    "@truffle/hdwallet-provider": "^1.2",
+    "apollo-fetch": "^0.7.0",
+    "babel-polyfill": "^6.26.0",
+    "babel-register": "^6.26.0",
+    "gluegun": "^4.6.1",
+    "truffle": "^5.2"
+  }
+}

tests/integration-tests/dynamic-data-source/schema.graphql

@@ -0,0 +1,4 @@
+type Foo @entity {
+  id: ID!
+  value: String!
+}

tests/integration-tests/dynamic-data-source/src/mapping.ts

@@ -0,0 +1,16 @@
+import { Trigger } from "../generated/Contract/Contract";
+import {Foo} from "../generated/schema";
+
+
+export function handleTrigger(event: Trigger): void {
+  let id = `${event.block.hash.toHexString()}${event.address.toHexString()}`;
+  let foo = new Foo(id);
+  foo.save();
+}
+
+
+
+
+
+
+

tests/integration-tests/dynamic-data-source/subgraph.yaml

@@ -0,0 +1,42 @@
+specVersion: 0.0.4
+schema:
+  file: ./schema.graphql
+dataSources:
+  - kind: ethereum/contract
+    name: Contract
+    network: test
+    source:
+      address: "0xCfEB869F69431e42cdB54A4F4f105C19C080A601"
+      abi: Contract
+    mapping:
+      kind: ethereum/events
+      apiVersion: 0.0.6
+      language: wasm/assemblyscript
+      abis:
+        - name: Contract
+          file: ./abis/Contract.abi
+      entities:
+        - Call
+      eventHandlers:
+        - event: Trigger(uint16)
+          handler: handleTrigger
+      file: ./src/mapping.ts
+templates:
+  - kind: ethereum/contract
+    name: Dynamic
+    network: test
+    source:
+      abi: Contract
+    mapping:
+      kind: ethereum/events
+      apiVersion: 0.0.6
+      language: wasm/assemblyscript
+      abis:
+        - name: Contract
+          file: ./abis/Contract.abi
+      entities:
+        - Call
+      eventHandlers:
+        - event: Trigger(uint16)
+          handler: handleTrigger
+      file: ./src/mapping.ts

tests/integration-tests/package.json

@@ -12,6 +12,7 @@
     "remove-then-update",
     "typename",
     "value-roundtrip",
+    "dynamic-data-source",
     "file-data-sources"
   ]
 }

tests/src/fixture.rs

@@ -20,14 +20,15 @@ use graph::cheap_clone::CheapClone;
 use graph::components::store::{BlockStore, DeploymentLocator};
 use graph::data::graphql::effort::LoadManager;
 use graph::data::query::{Query, QueryTarget};
-use graph::env::ENV_VARS;
+use graph::env::EnvVars;
 use graph::ipfs_client::IpfsClient;
 use graph::prelude::ethabi::ethereum_types::H256;
 use graph::prelude::{
     async_trait, r, ApiVersion, BlockNumber, DeploymentHash, GraphQlRunner as _, LoggerFactory,
     MetricsRegistry, NodeId, QueryError, SubgraphAssignmentProvider, SubgraphName,
     SubgraphRegistrar, SubgraphStore as _, SubgraphVersionSwitchingMode,
 };
+use graph::slog::crit;
 use graph_core::polling_monitor::ipfs_service::IpfsService;
 use graph_core::{
     LinkResolver, SubgraphAssignmentProvider as IpfsSubgraphAssignmentProvider,
@@ -38,7 +39,7 @@ use graph_mock::MockMetricsRegistry;
 use graph_node::manager::PanicSubscriptionManager;
 use graph_node::{config::Config, store_builder::StoreBuilder};
 use graph_store_postgres::{ChainHeadUpdateListener, ChainStore, Store, SubgraphStore};
-use slog::{crit, info, Logger};
+use slog::{info, Logger};
 use std::env::VarError;
 use std::pin::Pin;
 use std::sync::Arc;
@@ -216,7 +217,13 @@ pub async fn setup<C: Blockchain>(
     stores: &Stores,
     chain: Arc<C>,
     graft_block: Option<BlockPtr>,
+    env_vars: Option<EnvVars>,
 ) -> TestContext {
+    let env_vars = match env_vars {
+        Some(ev) => ev,
+        None => EnvVars::from_env().unwrap(),
+    };
+
     let logger = graph::log::logger(true);
     let logger_factory = LoggerFactory::new(logger.clone(), None);
     let mock_registry: Arc<dyn MetricsRegistry> = Arc::new(MockMetricsRegistry::new());
@@ -229,7 +236,7 @@ pub async fn setup<C: Blockchain>(
     let mut blockchain_map = BlockchainMap::new();
     blockchain_map.insert(stores.network_name.clone(), chain);
 
-    let static_filters = ENV_VARS.experimental_static_filters;
+    let static_filters = env_vars.experimental_static_filters;
 
     let ipfs = IpfsClient::localhost();
     let link_resolver = Arc::new(LinkResolver::new(
@@ -238,9 +245,9 @@ pub async fn setup<C: Blockchain>(
     ));
     let ipfs_service = IpfsService::new(
         ipfs,
-        ENV_VARS.mappings.max_ipfs_file_bytes as u64,
-        ENV_VARS.mappings.ipfs_timeout,
-        ENV_VARS.mappings.max_ipfs_concurrent_requests,
+        env_vars.mappings.max_ipfs_file_bytes as u64,
+        env_vars.mappings.ipfs_timeout,
+        env_vars.mappings.max_ipfs_concurrent_requests,
     );
 
     let blockchain_map = Arc::new(blockchain_map);

tests/tests/runner.rs

@@ -2,6 +2,7 @@ use std::sync::Arc;
 
 use cid::Cid;
 use graph::blockchain::{Block, BlockPtr};
+use graph::env::EnvVars;
 use graph::object;
 use graph::prelude::ethabi::ethereum_types::H256;
 use graph::prelude::{SubgraphAssignmentProvider, SubgraphName};
@@ -33,7 +34,15 @@ async fn data_source_revert() -> anyhow::Result<()> {
     };
 
     let chain = Arc::new(chain(blocks.clone(), &stores).await);
-    let ctx = fixture::setup(subgraph_name.clone(), &hash, &stores, chain.clone(), None).await;
+    let ctx = fixture::setup(
+        subgraph_name.clone(),
+        &hash,
+        &stores,
+        chain.clone(),
+        None,
+        None,
+    )
+    .await;
 
     let stop_block = test_ptr(2);
     ctx.start_and_sync_to(stop_block).await;
@@ -51,7 +60,15 @@ async fn data_source_revert() -> anyhow::Result<()> {
     )
     .await;
     let graft_block = Some(test_ptr(3));
-    let ctx = fixture::setup(subgraph_name.clone(), &hash, &stores, chain, graft_block).await;
+    let ctx = fixture::setup(
+        subgraph_name.clone(),
+        &hash,
+        &stores,
+        chain,
+        graft_block,
+        None,
+    )
+    .await;
     let stop_block = test_ptr(4);
     ctx.start_and_sync_to(stop_block).await;
 
@@ -97,7 +114,7 @@ async fn typename() -> anyhow::Result<()> {
 
     let stores = stores("./integration-tests/config.simple.toml").await;
     let chain = Arc::new(chain(blocks, &stores).await);
-    let ctx = fixture::setup(subgraph_name.clone(), &hash, &stores, chain, None).await;
+    let ctx = fixture::setup(subgraph_name.clone(), &hash, &stores, chain, None, None).await;
 
     ctx.start_and_sync_to(stop_block).await;
 
@@ -122,7 +139,7 @@ async fn file_data_sources() {
     };
     let stop_block = test_ptr(1);
     let chain = Arc::new(chain(blocks, &stores).await);
-    let ctx = fixture::setup(subgraph_name.clone(), &hash, &stores, chain, None).await;
+    let ctx = fixture::setup(subgraph_name.clone(), &hash, &stores, chain, None, None).await;
     ctx.start_and_sync_to(stop_block).await;
 
     // CID QmVkvoPGi9jvvuxsHDVJDgzPEzagBaWSZRYoRDzU244HjZ is the file
@@ -151,3 +168,56 @@ async fn file_data_sources() {
     let stop_block = test_ptr(2);
     ctx.start_and_sync_to(stop_block).await;
 }
+
+#[tokio::test]
+async fn template_static_filters_false_positives() {
+    let stores = stores("./integration-tests/config.simple.toml").await;
+
+    let subgraph_name = SubgraphName::new("dynamic-data-source").unwrap();
+    let hash = {
+        let test_dir = format!("./integration-tests/{}", subgraph_name);
+        fixture::build_subgraph(&test_dir).await
+    };
+
+    let blocks = {
+        let block_0 = genesis();
+        let block_1 = empty_block(block_0.ptr(), test_ptr(1));
+        let block_2 = empty_block(block_1.ptr(), test_ptr(2));
+        vec![block_0, block_1, block_2]
+    };
+    let stop_block = test_ptr(1);
+    let chain = Arc::new(chain(blocks, &stores).await);
+
+    let mut env_vars = EnvVars::default();
+    env_vars.experimental_static_filters = true;
+
+    let ctx = fixture::setup(
+        subgraph_name.clone(),
+        &hash,
+        &stores,
+        chain,
+        None,
+        Some(env_vars),
+    )
+    .await;
+    ctx.start_and_sync_to(stop_block).await;
+
+    let poi = ctx
+        .store
+        .get_proof_of_indexing(&ctx.deployment.hash, &None, test_ptr(1))
+        .await
+        .unwrap();
+
+    // This check exists to prevent regression of https://github.com/graphprotocol/graph-node/issues/3963
+    // when false positives go through the block stream, they should be discarded by
+    // `DataSource::match_and_decode`. The POI below is generated consistently from the empty
+    // POI table. If this fails it's likely that either the bug was re-introduced or there is
+    // a change in the POI infrastructure.
+    assert_eq!(
+        poi.unwrap(),
+        [
+            196, 173, 167, 52, 226, 19, 154, 61, 189, 94, 19, 229, 18, 7, 0, 252, 234, 49, 110,
+            179, 105, 64, 16, 46, 25, 194, 83, 94, 195, 225, 56, 252
+        ],
+    );
+}