If subgraph block handlers exist force realtime processing in watcher (#524)

* Avoid historical sync for block handlers and eth_getLogs without event handlers

* Stop processing after endBlock in watcher fill CLI

* Codegen changes for avoiding eth_getLogs when no subgraph event handlers exist

* Upgrade package version to 0.2.100

* Check for blockHandlerExists first before block processing

* Fix comments for historical block processing checks

lerna.json

@@ -2,7 +2,7 @@
   "packages": [
     "packages/*"
   ],
-  "version": "0.2.99",
+  "version": "0.2.100",
   "npmClient": "yarn",
   "useWorkspaces": true,
   "command": {

packages/cache/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/cache",
-  "version": "0.2.99",
+  "version": "0.2.100",
   "description": "Generic object cache",
   "main": "dist/index.js",
   "scripts": {

packages/cli/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/cli",
-  "version": "0.2.99",
+  "version": "0.2.100",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "scripts": {
@@ -15,13 +15,13 @@
   },
   "dependencies": {
     "@apollo/client": "^3.7.1",
-    "@cerc-io/cache": "^0.2.99",
+    "@cerc-io/cache": "^0.2.100",
-    "@cerc-io/ipld-eth-client": "^0.2.99",
+    "@cerc-io/ipld-eth-client": "^0.2.100",
     "@cerc-io/libp2p": "^0.42.2-laconic-0.1.4",
     "@cerc-io/nitro-node": "^0.1.15",
-    "@cerc-io/peer": "^0.2.99",
+    "@cerc-io/peer": "^0.2.100",
-    "@cerc-io/rpc-eth-client": "^0.2.99",
+    "@cerc-io/rpc-eth-client": "^0.2.100",
-    "@cerc-io/util": "^0.2.99",
+    "@cerc-io/util": "^0.2.100",
     "@ethersproject/providers": "^5.4.4",
     "@graphql-tools/utils": "^9.1.1",
     "@ipld/dag-cbor": "^8.0.0",

packages/codegen/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/codegen",
-  "version": "0.2.99",
+  "version": "0.2.100",
   "description": "Code generator",
   "private": true,
   "main": "index.js",
@@ -20,7 +20,7 @@
   },
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
-    "@cerc-io/util": "^0.2.99",
+    "@cerc-io/util": "^0.2.100",
     "@graphql-tools/load-files": "^6.5.2",
     "@npmcli/package-json": "^5.0.0",
     "@poanet/solidity-flattener": "https://github.com/vulcanize/solidity-flattener.git",

packages/codegen/src/templates/config-template.handlebars

@@ -10,8 +10,7 @@
   checkpointInterval = 2000
 
   # Enable state creation
-  # CAUTION: Disable only if state creation is not desired or can be filled subsequently
+  enableState = false
-  enableState = true
 
   {{#if (subgraphPath)}}
   subgraphPath = "./subgraph-build"

packages/codegen/src/templates/indexer-template.handlebars

@@ -829,6 +829,7 @@ export class Indexer implements IndexerInterface {
   {{/each}}
   }
 
+  // eslint-disable-next-line @typescript-eslint/no-empty-function
   _populateRelationsMap (): void {
   {{#each subgraphEntities as | subgraphEntity |}}
     {{#if subgraphEntity.relations}}
@@ -867,7 +868,27 @@ export class Indexer implements IndexerInterface {
     assert(blockHash);
     assert(blockNumber);
 
-    const { events: dbEvents, transactions } = await this._baseIndexer.fetchEvents(blockHash, blockNumber, this.eventSignaturesMap, this.parseEventNameAndArgs.bind(this));
+    {{#if (subgraphPath)}}
+    let dbEvents: DeepPartial<Event>[] = [];
+    let transactions: EthFullTransaction[] = [];
+
+    // Fetch events and txs only if subgraph config has any event handlers
+    if (this._graphWatcher.eventHandlerExists) {
+      ({ events: dbEvents, transactions } = await this._baseIndexer.fetchEvents(
+        blockHash,
+        blockNumber,
+        this.eventSignaturesMap,
+        this.parseEventNameAndArgs.bind(this)
+      ));
+    }
+    {{else~}}
+    const { events: dbEvents, transactions } = await this._baseIndexer.fetchEvents(
+      blockHash,
+      blockNumber,
+      this.eventSignaturesMap,
+      this.parseEventNameAndArgs.bind(this)
+    );
+    {{/if}}
 
     const dbTx = await this._db.createTransactionRunner();
     try {

packages/codegen/src/templates/package-template.handlebars

@@ -41,12 +41,12 @@
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
     "@apollo/client": "^3.3.19",
-    "@cerc-io/cli": "^0.2.99",
+    "@cerc-io/cli": "^0.2.100",
-    "@cerc-io/ipld-eth-client": "^0.2.99",
+    "@cerc-io/ipld-eth-client": "^0.2.100",
-    "@cerc-io/solidity-mapper": "^0.2.99",
+    "@cerc-io/solidity-mapper": "^0.2.100",
-    "@cerc-io/util": "^0.2.99",
+    "@cerc-io/util": "^0.2.100",
     {{#if (subgraphPath)}}
-    "@cerc-io/graph-node": "^0.2.99",
+    "@cerc-io/graph-node": "^0.2.100",
     {{/if}}
     "@ethersproject/providers": "^5.4.4",
     "debug": "^4.3.1",

packages/graph-node/package.json

@@ -1,10 +1,10 @@
 {
   "name": "@cerc-io/graph-node",
-  "version": "0.2.99",
+  "version": "0.2.100",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "devDependencies": {
-    "@cerc-io/solidity-mapper": "^0.2.99",
+    "@cerc-io/solidity-mapper": "^0.2.100",
     "@ethersproject/providers": "^5.4.4",
     "@graphprotocol/graph-ts": "^0.22.0",
     "@nomiclabs/hardhat-ethers": "^2.0.2",
@@ -51,9 +51,9 @@
   "dependencies": {
     "@apollo/client": "^3.3.19",
     "@cerc-io/assemblyscript": "0.19.10-watcher-ts-0.1.2",
-    "@cerc-io/cache": "^0.2.99",
+    "@cerc-io/cache": "^0.2.100",
-    "@cerc-io/ipld-eth-client": "^0.2.99",
+    "@cerc-io/ipld-eth-client": "^0.2.100",
-    "@cerc-io/util": "^0.2.99",
+    "@cerc-io/util": "^0.2.100",
     "@types/json-diff": "^0.5.2",
     "@types/yargs": "^17.0.0",
     "bn.js": "^4.11.9",

packages/graph-node/src/watcher.ts

@@ -60,6 +60,9 @@ export class GraphWatcher {
 
   _context: Context;
 
+  _blockHandlerExists = false;
+  _eventHandlerExists = false;
+
   constructor (database: GraphDatabase, ethClient: EthClient, ethProvider: providers.BaseProvider, serverConfig: ServerConfig) {
     this._database = database;
     this._ethClient = ethClient;
@@ -110,6 +113,10 @@ export class GraphWatcher {
       };
     }, {});
 
+    // Check if handlers exist for deciding watcher behaviour
+    this._blockHandlerExists = this._dataSources.some(dataSource => Boolean(dataSource.mapping.blockHandlers));
+    this._eventHandlerExists = this._dataSources.some(dataSource => Boolean(dataSource.mapping.eventHandlers));
+
     const data = await Promise.all(dataPromises);
 
     // Create a map from dataSource contract address to instance and contract interface.
@@ -151,6 +158,14 @@ export class GraphWatcher {
     return this._dataSources;
   }
 
+  get blockHandlerExists (): boolean {
+    return this._blockHandlerExists;
+  }
+
+  get eventHandlerExists (): boolean {
+    return this._eventHandlerExists;
+  }
+
   async addContracts () {
     assert(this._indexer);
     assert(this._indexer.watchContract);

packages/ipld-eth-client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/ipld-eth-client",
-  "version": "0.2.99",
+  "version": "0.2.100",
   "description": "IPLD ETH Client",
   "main": "dist/index.js",
   "scripts": {
@@ -20,8 +20,8 @@
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
     "@apollo/client": "^3.7.1",
-    "@cerc-io/cache": "^0.2.99",
+    "@cerc-io/cache": "^0.2.100",
-    "@cerc-io/util": "^0.2.99",
+    "@cerc-io/util": "^0.2.100",
     "cross-fetch": "^3.1.4",
     "debug": "^4.3.1",
     "ethers": "^5.4.4",

packages/peer/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/peer",
-  "version": "0.2.99",
+  "version": "0.2.100",
   "description": "libp2p module",
   "main": "dist/index.js",
   "exports": "./dist/index.js",

packages/rpc-eth-client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/rpc-eth-client",
-  "version": "0.2.99",
+  "version": "0.2.100",
   "description": "RPC ETH Client",
   "main": "dist/index.js",
   "scripts": {
@@ -19,9 +19,9 @@
   },
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
-    "@cerc-io/cache": "^0.2.99",
+    "@cerc-io/cache": "^0.2.100",
-    "@cerc-io/ipld-eth-client": "^0.2.99",
+    "@cerc-io/ipld-eth-client": "^0.2.100",
-    "@cerc-io/util": "^0.2.99",
+    "@cerc-io/util": "^0.2.100",
     "chai": "^4.3.4",
     "ethers": "^5.4.4",
     "left-pad": "^1.3.0",

packages/solidity-mapper/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/solidity-mapper",
-  "version": "0.2.99",
+  "version": "0.2.100",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "devDependencies": {

packages/test/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/test",
-  "version": "0.2.99",
+  "version": "0.2.100",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "private": true,

packages/tracing-client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/tracing-client",
-  "version": "0.2.99",
+  "version": "0.2.100",
   "description": "ETH VM tracing client",
   "main": "dist/index.js",
   "scripts": {

packages/util/package.json

@@ -1,13 +1,13 @@
 {
   "name": "@cerc-io/util",
-  "version": "0.2.99",
+  "version": "0.2.100",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "dependencies": {
     "@apollo/utils.keyvaluecache": "^1.0.1",
     "@cerc-io/nitro-node": "^0.1.15",
-    "@cerc-io/peer": "^0.2.99",
+    "@cerc-io/peer": "^0.2.100",
-    "@cerc-io/solidity-mapper": "^0.2.99",
+    "@cerc-io/solidity-mapper": "^0.2.100",
     "@cerc-io/ts-channel": "1.0.3-ts-nitro-0.1.1",
     "@ethersproject/properties": "^5.7.0",
     "@ethersproject/providers": "^5.4.4",
@@ -54,7 +54,7 @@
     "yargs": "^17.0.1"
   },
   "devDependencies": {
-    "@cerc-io/cache": "^0.2.99",
+    "@cerc-io/cache": "^0.2.100",
     "@nomiclabs/hardhat-waffle": "^2.0.1",
     "@types/bunyan": "^1.8.8",
     "@types/express": "^4.17.14",

packages/util/src/events.ts

@@ -123,9 +123,15 @@ export class EventWatcher {
       startBlockNumber = syncStatus.chainHeadBlockNumber + 1;
     }
 
-    // Check if filter for logs is enabled
+    // Perform checks before starting historical block processing
-    // Check if starting block for watcher is before latest canonical block
+    if (
-    if (this._config.jobQueue.useBlockRanges && startBlockNumber < latestCanonicalBlockNumber) {
+      // Skip historical block processing if any block handler exists
+      !this._indexer.graphWatcher?.blockHandlerExists &&
+      // Run historical block processing if useBlockRanges is enabled
+      this._config.jobQueue.useBlockRanges &&
+      // Only run historical block processing if we are below the frothy region
+      startBlockNumber < latestCanonicalBlockNumber
+    ) {
       await this.startHistoricalBlockProcessing(startBlockNumber, latestCanonicalBlockNumber);
 
       return;

packages/util/src/fill.ts

@@ -80,12 +80,12 @@ export const fillBlocks = async (
       const completePercentage = Math.round(blocksProcessed / numberOfBlocks * 100);
       log(`Processed ${blocksProcessed} of ${numberOfBlocks} blocks (${completePercentage}%)`);
 
-      await processBlockByNumber(jobQueue, blockNumber + 1);
+      if (blockNumber + 1 > endBlock) {
-
+        // Break the async loop when next block to be processed is more than endBlock.
-      if (blockNumber + 1 >= endBlock) {
-        // Break the async loop when blockProgress event is for the endBlock and processing is complete.
         break;
       }
+
+      await processBlockByNumber(jobQueue, blockNumber + 1);
     }
   }
 

packages/util/src/types.ts

@@ -288,6 +288,8 @@ export interface GraphDatabaseInterface {
 }
 
 export interface GraphWatcherInterface {
+  readonly blockHandlerExists: boolean;
+  readonly eventHandlerExists: boolean;
   init (): Promise<void>;
   setIndexer (indexer: IndexerInterface): void;
 }