Fix sync status update in historical processing job (#488)

* Fill watcher start block only in active mode

* Fix sync status update in historical processing

* Upgrade package versions

* Upgrade package versions in codegen template

* Fix graph-node test indexer

lerna.json

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

packages/cache/package.json

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

packages/cli/package.json

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

packages/cli/src/server.ts

@@ -285,25 +285,26 @@ export class ServerCmd {
     assert(indexer);
     assert(eventWatcher);
 
-    const syncStatus = await indexer.getSyncStatus();
-    if (!syncStatus) {
-      const contracts = await this.database.getContracts();
-      const startBlock = getStartBlock(contracts);
-      await fillBlocks(
-        jobQueue,
-        indexer,
-        eventWatcher,
-        config.jobQueue.blockDelayInMilliSecs,
-        {
-          startBlock,
-          endBlock: startBlock
-        }
-      );
-    }
-
     if (config.server.kind === KIND_ACTIVE) {
       // Delete all active and pending (before completed) jobs to prevent creating jobs after completion of processing previous block
       await jobQueue.deleteAllJobs('completed');
+
+      const syncStatus = await indexer.getSyncStatus();
+      if (!syncStatus) {
+        const contracts = await this.database.getContracts();
+        const startBlock = getStartBlock(contracts);
+        await fillBlocks(
+          jobQueue,
+          indexer,
+          eventWatcher,
+          config.jobQueue.blockDelayInMilliSecs,
+          {
+            startBlock,
+            endBlock: startBlock
+          }
+        );
+      }
+
       await eventWatcher.start();
     }
 

packages/codegen/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/codegen",
-  "version": "0.2.74",
+  "version": "0.2.75",
   "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.74",
+    "@cerc-io/util": "^0.2.75",
     "@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/database-template.handlebars

@@ -265,6 +265,12 @@ export class Database implements DatabaseInterface {
     return this._baseDatabase.updateSyncStatusIndexingError(repo, hasIndexingError);
   }
 
+  async updateSyncStatus (queryRunner: QueryRunner, syncStatus: DeepPartial<SyncStatus>): Promise<SyncStatus> {
+    const repo = queryRunner.manager.getRepository(SyncStatus);
+
+    return this._baseDatabase.updateSyncStatus(repo, syncStatus);
+  }
+
   async getSyncStatus (queryRunner: QueryRunner): Promise<SyncStatus | undefined> {
     const repo = queryRunner.manager.getRepository(SyncStatus);
 

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

@@ -673,6 +673,10 @@ export class Indexer implements IndexerInterface {
     return this._baseIndexer.updateSyncStatusIndexingError(hasIndexingError);
   }
 
+  async updateSyncStatus (syncStatus: DeepPartial<SyncStatus>): Promise<SyncStatus> {
+    return this._baseIndexer.updateSyncStatus(syncStatus);
+  }
+
   async getEvent (id: string): Promise<Event | undefined> {
     return this._baseIndexer.getEvent(id);
   }

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.74",
+    "@cerc-io/cli": "^0.2.75",
-    "@cerc-io/ipld-eth-client": "^0.2.74",
+    "@cerc-io/ipld-eth-client": "^0.2.75",
-    "@cerc-io/solidity-mapper": "^0.2.74",
+    "@cerc-io/solidity-mapper": "^0.2.75",
-    "@cerc-io/util": "^0.2.74",
+    "@cerc-io/util": "^0.2.75",
     {{#if (subgraphPath)}}
-    "@cerc-io/graph-node": "^0.2.74",
+    "@cerc-io/graph-node": "^0.2.75",
     {{/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.74",
+  "version": "0.2.75",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "devDependencies": {
-    "@cerc-io/solidity-mapper": "^0.2.74",
+    "@cerc-io/solidity-mapper": "^0.2.75",
     "@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.74",
+    "@cerc-io/cache": "^0.2.75",
-    "@cerc-io/ipld-eth-client": "^0.2.74",
+    "@cerc-io/ipld-eth-client": "^0.2.75",
-    "@cerc-io/util": "^0.2.74",
+    "@cerc-io/util": "^0.2.75",
     "@types/json-diff": "^0.5.2",
     "@types/yargs": "^17.0.0",
     "bn.js": "^4.11.9",

packages/graph-node/test/utils/indexer.ts

@@ -190,6 +190,12 @@ export class Indexer implements IndexerInterface {
     return undefined;
   }
 
+  async updateSyncStatus (syncStatus: SyncStatusInterface): Promise<SyncStatusInterface> {
+    assert(syncStatus);
+
+    return {} as SyncStatusInterface;
+  }
+
   async markBlocksAsPruned (blocks: BlockProgressInterface[]): Promise<void> {
     assert(blocks);
 

packages/ipld-eth-client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/ipld-eth-client",
-  "version": "0.2.74",
+  "version": "0.2.75",
   "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.74",
+    "@cerc-io/cache": "^0.2.75",
-    "@cerc-io/util": "^0.2.74",
+    "@cerc-io/util": "^0.2.75",
     "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.74",
+  "version": "0.2.75",
   "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.74",
+  "version": "0.2.75",
   "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.74",
+    "@cerc-io/cache": "^0.2.75",
-    "@cerc-io/ipld-eth-client": "^0.2.74",
+    "@cerc-io/ipld-eth-client": "^0.2.75",
-    "@cerc-io/util": "^0.2.74",
+    "@cerc-io/util": "^0.2.75",
     "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.74",
+  "version": "0.2.75",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "devDependencies": {

packages/test/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/test",
-  "version": "0.2.74",
+  "version": "0.2.75",
   "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.74",
+  "version": "0.2.75",
   "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.74",
+  "version": "0.2.75",
   "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.74",
+    "@cerc-io/peer": "^0.2.75",
-    "@cerc-io/solidity-mapper": "^0.2.74",
+    "@cerc-io/solidity-mapper": "^0.2.75",
     "@cerc-io/ts-channel": "1.0.3-ts-nitro-0.1.1",
     "@ethersproject/properties": "^5.7.0",
     "@ethersproject/providers": "^5.4.4",
@@ -52,7 +52,7 @@
     "yargs": "^17.0.1"
   },
   "devDependencies": {
-    "@cerc-io/cache": "^0.2.74",
+    "@cerc-io/cache": "^0.2.75",
     "@nomiclabs/hardhat-waffle": "^2.0.1",
     "@types/bunyan": "^1.8.8",
     "@types/express": "^4.17.14",

packages/util/src/database.ts

@@ -211,6 +211,15 @@ export class Database {
     return repo.save(entity);
   }
 
+  async updateSyncStatus (repo: Repository<SyncStatusInterface>, syncStatus: DeepPartial<SyncStatusInterface>): Promise<SyncStatusInterface> {
+    const entity = await repo.findOne();
+
+    return await repo.save({
+      ...entity,
+      ...syncStatus
+    });
+  }
+
   async getBlockProgress (repo: Repository<BlockProgressInterface>, blockHash: string): Promise<BlockProgressInterface | undefined> {
     return repo.findOne({ where: { blockHash } });
   }

packages/util/src/indexer.ts

@@ -291,6 +291,23 @@ export class Indexer {
     return res;
   }
 
+  async updateSyncStatus (syncStatus: DeepPartial<SyncStatusInterface>): Promise<SyncStatusInterface> {
+    const dbTx = await this._db.createTransactionRunner();
+    let res;
+
+    try {
+      res = await this._db.updateSyncStatus(dbTx, syncStatus);
+      await dbTx.commitTransaction();
+    } catch (error) {
+      await dbTx.rollbackTransaction();
+      throw error;
+    } finally {
+      await dbTx.release();
+    }
+
+    return res;
+  }
+
   async getBlocks (blockFilter: { blockNumber?: number, blockHash?: string }): Promise<Array<EthFullBlock | null>> {
     assert(blockFilter.blockHash || blockFilter.blockNumber);
     const blocks = await this._ethClient.getFullBlocks(blockFilter);

packages/util/src/job-runner.ts

@@ -237,11 +237,14 @@ export class JobRunner {
 
     // Update sync status canonical, indexed and chain head block to end block
     // Update with zero hash as they won't be used during historical processing
-    await Promise.all([
+    await this._indexer.updateSyncStatus({
-      this._indexer.updateSyncStatusCanonicalBlock(constants.HashZero, endBlock, true),
+      latestCanonicalBlockHash: constants.HashZero,
-      this._indexer.updateSyncStatusIndexedBlock(constants.HashZero, endBlock, true),
+      latestIndexedBlockHash: constants.HashZero,
-      this._indexer.updateSyncStatusChainHead(constants.HashZero, endBlock, true)
+      chainHeadBlockHash: constants.HashZero,
-    ]);
+      latestCanonicalBlockNumber: endBlock,
+      latestIndexedBlockNumber: endBlock,
+      chainHeadBlockNumber: endBlock
+    });
     log(`Sync status canonical, indexed and chain head block updated to ${endBlock}`);
 
     this._historicalProcessingCompletedUpto = endBlock;

packages/util/src/types.ts

@@ -195,6 +195,7 @@ export interface IndexerInterface {
   updateSyncStatusProcessedBlock (blockHash: string, blockNumber: number, force?: boolean): Promise<SyncStatusInterface>
   updateStateSyncStatusIndexedBlock (blockNumber: number, force?: boolean): Promise<StateSyncStatusInterface | undefined>
   updateStateSyncStatusCheckpointBlock (blockNumber: number, force?: boolean): Promise<StateSyncStatusInterface>
+  updateSyncStatus (syncStatus: DeepPartial<SyncStatusInterface>): Promise<SyncStatusInterface>;
   markBlocksAsPruned (blocks: BlockProgressInterface[]): Promise<void>
   saveEventEntity (dbEvent: EventInterface): Promise<EventInterface>
   saveEvents (dbEvents: DeepPartial<EventInterface>[]): Promise<void>
@@ -258,6 +259,7 @@ export interface DatabaseInterface {
   updateSyncStatusCanonicalBlock (queryRunner: QueryRunner, blockHash: string, blockNumber: number, force?: boolean): Promise<SyncStatusInterface>;
   updateSyncStatusIndexingError (queryRunner: QueryRunner, hasIndexingError: boolean): Promise<SyncStatusInterface | undefined>;
   updateSyncStatusProcessedBlock (queryRunner: QueryRunner, blockHash: string, blockNumber: number, force?: boolean): Promise<SyncStatusInterface>;
+  updateSyncStatus (queryRunner: QueryRunner, syncStatus: DeepPartial<SyncStatusInterface>): Promise<SyncStatusInterface>;
   saveEvents (queryRunner: QueryRunner, events: DeepPartial<EventInterface>[]): Promise<void>;
   saveBlockWithEvents (queryRunner: QueryRunner, block: DeepPartial<BlockProgressInterface>, events: DeepPartial<EventInterface>[]): Promise<BlockProgressInterface>;
   saveEventEntity (queryRunner: QueryRunner, entity: EventInterface): Promise<EventInterface>;