Fix block processing job created twice when processing missing parent block (#67)

* Fix block processing job created twice in watcher

* Fix block processing job for missing parent blocks

packages/eden-watcher/src/entity/BlockProgress.ts

@@ -2,7 +2,7 @@
 // Copyright 2021 Vulcanize, Inc.
 //
 
-import { Entity, PrimaryGeneratedColumn, Column, Index } from 'typeorm';
+import { Entity, PrimaryGeneratedColumn, Column, Index, CreateDateColumn } from 'typeorm';
 import { BlockProgressInterface } from '@vulcanize/util';
 
 @Entity()
@@ -42,4 +42,7 @@ export class BlockProgress implements BlockProgressInterface {
 
   @Column('boolean', { default: false })
   isPruned!: boolean;
+
+  @CreateDateColumn()
+  createdAt!: Date;
 }

packages/graph-node/src/loader.ts

@@ -159,7 +159,6 @@ export const instantiate = async (
           // TODO: Check for function overloading.
           let result = await contract[functionName](...functionParams);
 
-          // TODO: Check for function overloading.
           // Using function signature does not work.
           const { outputs } = contract.interface.getFunction(functionName);
           assert(outputs);

packages/graph-node/test/subgraph/example1/src/mapping.ts

@@ -143,7 +143,6 @@ export function testStructEthCall (): void {
   log.debug('In test struct eth call', []);
 
   // Bind the contract to the address that emitted the event.
-  // TODO: Address.fromString throws error in WASM module instantiation.
   const contractAddress = dataSource.address();
   const contract = Example1.bind(contractAddress);
 

packages/graph-test-watcher/src/entity/BlockProgress.ts

@@ -2,7 +2,7 @@
 // Copyright 2021 Vulcanize, Inc.
 //
 
-import { Entity, PrimaryGeneratedColumn, Column, Index } from 'typeorm';
+import { Entity, PrimaryGeneratedColumn, Column, Index, CreateDateColumn } from 'typeorm';
 import { BlockProgressInterface } from '@vulcanize/util';
 
 @Entity()
@@ -42,4 +42,7 @@ export class BlockProgress implements BlockProgressInterface {
 
   @Column('boolean', { default: false })
   isPruned!: boolean;
+
+  @CreateDateColumn()
+  createdAt!: Date;
 }

packages/util/src/events.ts

@@ -50,7 +50,7 @@ export class EventWatcher {
 
   async startBlockProcessing (): Promise<void> {
     const syncStatus = await this._indexer.getSyncStatus();
-    let startBlockNumber;
+    let startBlockNumber: number;
 
     if (!syncStatus) {
       // Get latest block in chain.
@@ -62,7 +62,8 @@ export class EventWatcher {
 
     const { ethServer: { blockDelayInMilliSecs } } = this._upstreamConfig;
 
-    processBlockByNumber(this._jobQueue, this._indexer, blockDelayInMilliSecs, startBlockNumber);
+    // Wait for block processing as blockProgress event might process the same block.
+    await processBlockByNumber(this._jobQueue, this._indexer, blockDelayInMilliSecs, startBlockNumber);
 
     // Creating an AsyncIterable from AsyncIterator to iterate over the values.
     // https://www.codementor.io/@tiagolopesferreira/asynchronous-iterators-in-javascript-jl1yg8la1#for-wait-of
@@ -141,13 +142,16 @@ export class EventWatcher {
 
   async _handleIndexingComplete (jobData: any): Promise<void> {
     const { blockHash, blockNumber, priority } = jobData;
-    log(`Job onComplete indexing block ${blockHash} ${blockNumber}`);
 
     const [blockProgress, syncStatus] = await Promise.all([
       this._indexer.getBlockProgress(blockHash),
+      // Update sync progress.
       this._indexer.updateSyncStatusIndexedBlock(blockHash, blockNumber)
     ]);
 
+    if (blockProgress) {
+      log(`Job onComplete indexing block ${blockHash} ${blockNumber}`);
+
       // Create pruning job if required.
       if (syncStatus && syncStatus.latestIndexedBlockNumber > (syncStatus.latestCanonicalBlockNumber + MAX_REORG_DEPTH)) {
         await createPruningJob(this._jobQueue, syncStatus.latestCanonicalBlockNumber, priority);
@@ -155,9 +159,12 @@ export class EventWatcher {
 
       // Publish block progress event if no events exist.
       // Event for blocks with events will be pusblished from eventProcessingCompleteHandler.
-    if (blockProgress && blockProgress.numEvents === 0) {
+      if (blockProgress.numEvents === 0) {
         await this.publishBlockProgressToSubscribers(blockProgress);
       }
+    } else {
+      log(`block not indexed for ${blockHash} ${blockNumber}`);
+    }
   }
 
   async _handlePruningComplete (jobData: any): Promise<void> {

packages/util/src/job-runner.ts

@@ -192,7 +192,7 @@ export class JobRunner {
         const message = `Parent block number ${parentBlockNumber} hash ${parentHash} of block number ${blockNumber} hash ${blockHash} not fetched yet, aborting`;
         log(message);
 
-        throw new Error(message);
+        return;
       }
 
       if (!parentBlock.isComplete) {