Fixes and improvements for eden-watcher job-runner and compare CLI (#165

)

* Compare IPLD state entity without derived fields

* Apply default limit to array relation fields in IPLD state entity

* Mark block as complete after processing of block handler

* Avoid re processing of block handler

* Use LIMIT 1 in the query to get latest IPLD block

* Replace eth_calls in eden-watcher with getStorageValue

* Add checkpoint verification to export state CLI

* Fix get diff blocks query when creating checkpoint

* Fix subgraph staker sort and remove entities sequentially in reset CLI

Co-authored-by: prathamesh0 <prathamesh.musale0@gmail.com>

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

@@ -102,10 +102,10 @@ export class Database implements IPLDDatabaseInterface {
   }
 
   // Fetch all diff IPLDBlocks after the specified block number.
-  async getDiffIPLDBlocksByBlocknumber (contractAddress: string, blockNumber: number): Promise<IPLDBlock[]> {
+  async getDiffIPLDBlocksInRange (contractAddress: string, startBlock: number, endBlock: number): Promise<IPLDBlock[]> {
     const repo = this._conn.getRepository(IPLDBlock);
 
-    return this._baseDatabase.getDiffIPLDBlocksByBlocknumber(repo, contractAddress, blockNumber);
+    return this._baseDatabase.getDiffIPLDBlocksInRange(repo, contractAddress, startBlock, endBlock);
   }
 
   async saveOrUpdateIPLDBlock (dbTx: QueryRunner, ipldBlock: IPLDBlock): Promise<IPLDBlock> {

packages/codegen/src/templates/export-state-template.handlebars

@@ -9,7 +9,17 @@ import debug from 'debug';
 import fs from 'fs';
 import path from 'path';
 
-import { Config, DEFAULT_CONFIG_PATH, getConfig, initClients, JobQueue, StateKind } from '@vulcanize/util';
+import {
+  Config,
+  DEFAULT_CONFIG_PATH,
+  getConfig,
+  initClients,
+  JobQueue,
+  {{#if (subgraphPath)}}
+  verifyCheckpointData,
+  {{/if}}
+  StateKind
+} from '@vulcanize/util';
 {{#if (subgraphPath)}}
 import { GraphWatcher, Database as GraphDatabase } from '@vulcanize/graph-node';
 {{/if}}
@@ -36,6 +46,20 @@ const main = async (): Promise<void> => {
       alias: 'o',
       type: 'string',
       describe: 'Export file path'
+    },
+    {{#if (subgraphPath)}}
+    verify: {
+      alias: 'v',
+      type: 'boolean',
+      describe: 'Verify checkpoint',
+      default: true
+    },
+    {{/if}}
+    createCheckpoint: {
+      alias: 'c',
+      type: 'boolean',
+      describe: 'Create new checkpoint',
+      default: false
     }
   }).argv;
 
@@ -98,13 +122,24 @@ const main = async (): Promise<void> => {
 
     // Create and export checkpoint if checkpointing is on for the contract.
     if (contract.checkpoint) {
-      await indexer.createCheckpoint(contract.address, block.blockHash);
+      if (argv.createCheckpoint) {
+        log(`Creating checkpoint at block ${block.blockNumber}`);
+        await indexer.createCheckpoint(contract.address, block.blockHash);
+      }
 
       const ipldBlock = await indexer.getLatestIPLDBlock(contract.address, StateKind.Checkpoint, block.blockNumber);
       assert(ipldBlock);
 
       const data = indexer.getIPLDData(ipldBlock);
 
+      {{#if (subgraphPath)}}
+      if (argv.verify) {
+        log(`Verifying checkpoint data for contract ${contract.address}`);
+        await verifyCheckpointData(graphDb, ipldBlock.block, data);
+        log('Checkpoint data verified');
+      }
+
+      {{/if}}
       if (indexer.isIPFSConfigured()) {
         await indexer.pushToIPFS(data);
       }

packages/eden-watcher/src/cli/export-state.ts

@@ -9,7 +9,7 @@ import debug from 'debug';
 import fs from 'fs';
 import path from 'path';
 
-import { Config, DEFAULT_CONFIG_PATH, getConfig, initClients, JobQueue, StateKind } from '@vulcanize/util';
+import { Config, DEFAULT_CONFIG_PATH, getConfig, initClients, JobQueue, StateKind, verifyCheckpointData } from '@vulcanize/util';
 import { GraphWatcher, Database as GraphDatabase } from '@vulcanize/graph-node';
 import * as codec from '@ipld/dag-cbor';
 
@@ -34,6 +34,18 @@ const main = async (): Promise<void> => {
       alias: 'o',
       type: 'string',
       describe: 'Export file path'
+    },
+    createCheckpoint: {
+      alias: 'c',
+      type: 'boolean',
+      describe: 'Create new checkpoint',
+      default: false
+    },
+    verify: {
+      alias: 'v',
+      type: 'boolean',
+      describe: 'Verify checkpoint',
+      default: true
     }
   }).argv;
 
@@ -92,13 +104,22 @@ const main = async (): Promise<void> => {
 
     // Create and export checkpoint if checkpointing is on for the contract.
     if (contract.checkpoint) {
-      await indexer.createCheckpoint(contract.address, block.blockHash);
+      if (argv.createCheckpoint) {
+        log(`Creating checkpoint at block ${block.blockNumber}`);
+        await indexer.createCheckpoint(contract.address, block.blockHash);
+      }
 
       const ipldBlock = await indexer.getLatestIPLDBlock(contract.address, StateKind.Checkpoint, block.blockNumber);
       assert(ipldBlock);
 
       const data = indexer.getIPLDData(ipldBlock);
 
+      if (argv.verify) {
+        log(`Verifying checkpoint data for contract ${contract.address}`);
+        await verifyCheckpointData(graphDb, ipldBlock.block, data);
+        log('Checkpoint data verified');
+      }
+
       if (indexer.isIPFSConfigured()) {
         await indexer.pushToIPFS(data);
       }

packages/eden-watcher/src/cli/reset-cmds/state.ts

@@ -83,11 +83,9 @@ export const handler = async (argv: any): Promise<void> => {
   try {
     const entities = [BlockProgress, Producer, ProducerSet, ProducerSetChange, ProducerRewardCollectorChange, RewardScheduleEntry, RewardSchedule, ProducerEpoch, Block, Epoch, SlotClaim, Slot, Staker, Network, Distributor, Distribution, Claim, Slash, Account];
 
-    const removeEntitiesPromise = entities.map(async entityClass => {
-      return db.removeEntities<any>(dbTx, entityClass, { blockNumber: MoreThan(argv.blockNumber) });
-    });
-
-    await Promise.all(removeEntitiesPromise);
+    for (const entity of entities) {
+      await db.removeEntities<any>(dbTx, entity, { blockNumber: MoreThan(argv.blockNumber) });
+    }
 
     const syncStatus = await indexer.getSyncStatus();
     assert(syncStatus, 'Missing syncStatus');

packages/eden-watcher/src/database.ts

@@ -62,10 +62,10 @@ export class Database implements IPLDDatabaseInterface {
   }
 
   // Fetch all diff IPLDBlocks after the specified block number.
-  async getDiffIPLDBlocksByBlocknumber (contractAddress: string, blockNumber: number): Promise<IPLDBlock[]> {
+  async getDiffIPLDBlocksInRange (contractAddress: string, startblock: number, endBlock: number): Promise<IPLDBlock[]> {
     const repo = this._conn.getRepository(IPLDBlock);
 
-    return this._baseDatabase.getDiffIPLDBlocksByBlocknumber(repo, contractAddress, blockNumber);
+    return this._baseDatabase.getDiffIPLDBlocksInRange(repo, contractAddress, startblock, endBlock);
   }
 
   async saveOrUpdateIPLDBlock (dbTx: QueryRunner, ipldBlock: IPLDBlock): Promise<IPLDBlock> {

packages/eden-watcher/src/hooks.ts

@@ -60,14 +60,14 @@ export async function createStateCheckpoint (indexer: Indexer, contractAddress:
 
   // Fetch the latest 'checkpoint' | 'init' for the contract to fetch diffs after it.
   let prevNonDiffBlock: IPLDBlockInterface;
-  let getDiffBlockNumber: number;
-  const checkpointBlock = await indexer.getLatestIPLDBlock(contractAddress, StateKind.Checkpoint, block.blockNumber);
+  let diffStartBlockNumber: number;
+  const checkpointBlock = await indexer.getLatestIPLDBlock(contractAddress, StateKind.Checkpoint, block.blockNumber - 1);
 
   if (checkpointBlock) {
     const checkpointBlockNumber = checkpointBlock.block.blockNumber;
 
     prevNonDiffBlock = checkpointBlock;
-    getDiffBlockNumber = checkpointBlockNumber;
+    diffStartBlockNumber = checkpointBlockNumber;
 
     // Update IPLD status map with the latest checkpoint info.
     // Essential while importing state as checkpoint at the snapshot block is added by import-state CLI.
@@ -80,11 +80,11 @@ export async function createStateCheckpoint (indexer: Indexer, contractAddress:
 
     prevNonDiffBlock = initBlock;
     // Take block number previous to initial state block to include any diff state at that block.
-    getDiffBlockNumber = initBlock.block.blockNumber - 1;
+    diffStartBlockNumber = initBlock.block.blockNumber - 1;
   }
 
   // Fetching all diff blocks after the latest 'checkpoint' | 'init'.
-  const diffBlocks = await indexer.getDiffIPLDBlocksByBlocknumber(contractAddress, getDiffBlockNumber);
+  const diffBlocks = await indexer.getDiffIPLDBlocksInRange(contractAddress, diffStartBlockNumber, block.blockNumber);
 
   const prevNonDiffBlockData = codec.decode(Buffer.from(prevNonDiffBlock.data)) as any;
   const data = {

packages/eden-watcher/src/indexer.ts

@@ -303,8 +303,8 @@ export class Indexer implements IPLDIndexerInterface {
     return this._baseIndexer.getIPLDBlockByCid(cid);
   }
 
-  async getDiffIPLDBlocksByBlocknumber (contractAddress: string, blockNumber: number): Promise<IPLDBlock[]> {
-    return this._db.getDiffIPLDBlocksByBlocknumber(contractAddress, blockNumber);
+  async getDiffIPLDBlocksInRange (contractAddress: string, startBlock: number, endBlock: number): Promise<IPLDBlock[]> {
+    return this._db.getDiffIPLDBlocksInRange(contractAddress, startBlock, endBlock);
   }
 
   getIPLDData (ipldBlock: IPLDBlock): any {

packages/erc20-watcher/src/cli/reset-cmds/state.ts

@@ -56,11 +56,9 @@ export const handler = async (argv: any): Promise<void> => {
   const dbTx = await db.createTransactionRunner();
 
   try {
-    const removeEntitiesPromise = [BlockProgress, Allowance, Balance].map(async entityClass => {
-      return db.removeEntities<any>(dbTx, entityClass, { blockNumber: MoreThan(argv.blockNumber) });
-    });
-
-    await Promise.all(removeEntitiesPromise);
+    for (const entity of [BlockProgress, Allowance, Balance]) {
+      await db.removeEntities<any>(dbTx, entity, { blockNumber: MoreThan(argv.blockNumber) });
+    }
 
     if (syncStatus.latestIndexedBlockNumber > blockProgress.blockNumber) {
       await indexer.updateSyncStatusIndexedBlock(blockProgress.blockHash, blockProgress.blockNumber, true);

packages/erc721-watcher/src/cli/export-state.ts

@@ -33,6 +33,12 @@ const main = async (): Promise<void> => {
       alias: 'o',
       type: 'string',
       describe: 'Export file path'
+    },
+    createCheckpoint: {
+      alias: 'c',
+      type: 'boolean',
+      describe: 'Create new checkpoint',
+      default: false
     }
   }).argv;
 
@@ -83,7 +89,10 @@ const main = async (): Promise<void> => {
 
     // Create and export checkpoint if checkpointing is on for the contract.
     if (contract.checkpoint) {
-      await indexer.createCheckpoint(contract.address, block.blockHash);
+      if (argv.createCheckpoint) {
+        log(`Creating checkpoint at block ${block.blockNumber}`);
+        await indexer.createCheckpoint(contract.address, block.blockHash);
+      }
 
       const ipldBlock = await indexer.getLatestIPLDBlock(contract.address, StateKind.Checkpoint, block.blockNumber);
       assert(ipldBlock);

packages/erc721-watcher/src/cli/reset-cmds/state.ts

@@ -70,11 +70,9 @@ export const handler = async (argv: any): Promise<void> => {
   try {
     const entities = [BlockProgress, SupportsInterface, BalanceOf, OwnerOf, GetApproved, IsApprovedForAll, Name, Symbol, TokenURI, _Name, _Symbol, _Owners, _Balances, _TokenApprovals, _OperatorApprovals];
 
-    const removeEntitiesPromise = entities.map(async entityClass => {
-      return db.removeEntities<any>(dbTx, entityClass, { blockNumber: MoreThan(argv.blockNumber) });
-    });
-
-    await Promise.all(removeEntitiesPromise);
+    for (const entity of entities) {
+      await db.removeEntities<any>(dbTx, entity, { blockNumber: MoreThan(argv.blockNumber) });
+    }
 
     const syncStatus = await indexer.getSyncStatus();
     assert(syncStatus, 'Missing syncStatus');

packages/erc721-watcher/src/database.ts

@@ -323,10 +323,10 @@ export class Database implements IPLDDatabaseInterface {
   }
 
   // Fetch all diff IPLDBlocks after the specified block number.
-  async getDiffIPLDBlocksByBlocknumber (contractAddress: string, blockNumber: number): Promise<IPLDBlock[]> {
+  async getDiffIPLDBlocksInRange (contractAddress: string, startBlock: number, endBlock: number): Promise<IPLDBlock[]> {
     const repo = this._conn.getRepository(IPLDBlock);
 
-    return this._baseDatabase.getDiffIPLDBlocksByBlocknumber(repo, contractAddress, blockNumber);
+    return this._baseDatabase.getDiffIPLDBlocksInRange(repo, contractAddress, startBlock, endBlock);
   }
 
   async saveOrUpdateIPLDBlock (dbTx: QueryRunner, ipldBlock: IPLDBlock): Promise<IPLDBlock> {

packages/graph-node/environments/compare-cli-config.toml

@@ -5,14 +5,14 @@
 [queries]
   queryDir = "../graph-test-watcher/src/gql/queries"
   names = []
-  idsEndpoint = "gqlEndpoint1"
   blockDelayInMs = 250
 
 [watcher]
-  configpath = "../../graph-test-watcher/environments/local.toml"
+  configPath = "../../graph-test-watcher/environments/local.toml"
   entitiesDir = "../../graph-test-watcher/src/entity"
   endpoint = "gqlEndpoint2"
   verifyState = true
+  derivedFields = []
 
 [cache]
   endpoint = "gqlEndpoint1"

packages/graph-node/src/cli/compare/compare-blocks.ts

@@ -9,11 +9,10 @@ import path from 'path';
 import assert from 'assert';
 import { SnakeNamingStrategy } from 'typeorm-naming-strategies';
 import _ from 'lodash';
-import omitDeep from 'omit-deep';
 import { getConfig as getWatcherConfig, wait } from '@vulcanize/util';
 import { GraphQLClient } from '@vulcanize/ipld-eth-client';
 
-import { compareObjects, compareQuery, Config, getBlockIPLDState as getIPLDStateByBlock, getClients, getConfig } from './utils';
+import { checkEntityInIPLDState, compareQuery, Config, getBlockIPLDState as getIPLDStateByBlock, getClients, getConfig } from './utils';
 import { Database } from '../../database';
 import { getSubgraphConfig } from '../../utils';
 
@@ -130,7 +129,12 @@ export const main = async (): Promise<void> => {
             );
 
             if (config.watcher.verifyState) {
-              await checkEntityInIPLDState(ipldStateByBlock, queryName, result, id, rawJson);
+              const ipldDiff = await checkEntityInIPLDState(ipldStateByBlock, queryName, result, id, rawJson, config.watcher.derivedFields);
+
+              if (ipldDiff) {
+                log('Results mismatch for IPLD state:', ipldDiff);
+                diffFound = true;
+              }
             }
 
             if (diff) {
@@ -167,22 +171,3 @@ export const main = async (): Promise<void> => {
     process.exit(1);
   }
 };
-
-const checkEntityInIPLDState = async (
-  ipldState: {[key: string]: any},
-  queryName: string,
-  entityResult: {[key: string]: any},
-  id: string,
-  rawJson: boolean
-) => {
-  const entityName = _.startCase(queryName);
-  const ipldEntity = ipldState[entityName][id];
-
-  // Filter __typename key in GQL result.
-  const resultEntity = omitDeep(entityResult[queryName], '__typename');
-  const diff = compareObjects(ipldEntity, resultEntity, rawJson);
-
-  if (diff) {
-    log('Results mismatch for IPLD state:', diff);
-  }
-};