Performance improvements for IPLDBlocks in eden watcher (#80)

* Avoid fetching contracts while creating a checkpoint

* Use enum for state kind and cache ipld status

* Avoid fetching block twice while finalizing a staged diff IPLDBlock

* Create checkpoints at fixed block numbers and refactor checkpointing code

* Avoid calling block handler until start block is reached

* Use delete while finalizing staged diff IPLDBlocks

* Add a check to ensure hooks job is created only once

* Avoid check for initial state while creating a checkpoint

packages/graph-test-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, STATE_KIND_CHECKPOINT } from '@vulcanize/util';
+import { Config, DEFAULT_CONFIG_PATH, getConfig, initClients, JobQueue, StateKind } from '@vulcanize/util';
 import { GraphWatcher, Database as GraphDatabase } from '@vulcanize/graph-node';
 import * as codec from '@ipld/dag-cbor';
 
@@ -94,7 +94,7 @@ const main = async (): Promise<void> => {
     if (contract.checkpoint) {
       await indexer.createCheckpoint(contract.address, block.blockHash);
 
-      const ipldBlock = await indexer.getLatestIPLDBlock(contract.address, STATE_KIND_CHECKPOINT, block.blockNumber);
+      const ipldBlock = await indexer.getLatestIPLDBlock(contract.address, StateKind.Checkpoint, block.blockNumber);
       assert(ipldBlock);
 
       const data = indexer.getIPLDData(ipldBlock);

packages/graph-test-watcher/src/cli/import-state.ts

@@ -11,7 +11,7 @@ import { PubSub } from 'apollo-server-express';
 import fs from 'fs';
 import path from 'path';
 
-import { getConfig, fillBlocks, JobQueue, DEFAULT_CONFIG_PATH, Config, initClients, STATE_KIND_INIT, STATE_KIND_DIFF_STAGED } from '@vulcanize/util';
+import { getConfig, fillBlocks, JobQueue, DEFAULT_CONFIG_PATH, Config, initClients, StateKind } from '@vulcanize/util';
 import { GraphWatcher, Database as GraphDatabase } from '@vulcanize/graph-node';
 import * as codec from '@ipld/dag-cbor';
 
@@ -112,8 +112,8 @@ export const main = async (): Promise<any> => {
   }
 
   // The 'diff_staged' and 'init' IPLD blocks are unnecessary as checkpoints have been already created for the snapshot block.
-  await indexer.removeIPLDBlocks(block.blockNumber, STATE_KIND_INIT);
-  await indexer.removeIPLDBlocks(block.blockNumber, STATE_KIND_DIFF_STAGED);
+  await indexer.removeIPLDBlocks(block.blockNumber, StateKind.Init);
+  await indexer.removeIPLDBlocks(block.blockNumber, StateKind.DiffStaged);
 };
 
 main().catch(err => {

packages/graph-test-watcher/src/database.ts

@@ -6,7 +6,7 @@ import assert from 'assert';
 import { Connection, ConnectionOptions, DeepPartial, FindConditions, QueryRunner, FindManyOptions } from 'typeorm';
 import path from 'path';
 
-import { IPLDDatabase as BaseDatabase, IPLDDatabaseInterface, QueryOptions, Where } from '@vulcanize/util';
+import { IPLDDatabase as BaseDatabase, IPLDDatabaseInterface, QueryOptions, StateKind, Where } from '@vulcanize/util';
 
 import { Contract } from './entity/Contract';
 import { Event } from './entity/Event';
@@ -83,7 +83,7 @@ export class Database implements IPLDDatabaseInterface {
     return this._baseDatabase.getIPLDBlocks(repo, where);
   }
 
-  async getLatestIPLDBlock (contractAddress: string, kind: string | null, blockNumber?: number): Promise<IPLDBlock | undefined> {
+  async getLatestIPLDBlock (contractAddress: string, kind: StateKind | null, blockNumber?: number): Promise<IPLDBlock | undefined> {
     const repo = this._conn.getRepository(IPLDBlock);
 
     return this._baseDatabase.getLatestIPLDBlock(repo, contractAddress, kind, blockNumber);
@@ -109,7 +109,9 @@ export class Database implements IPLDDatabaseInterface {
   }
 
   async removeIPLDBlocks (dbTx: QueryRunner, blockNumber: number, kind: string): Promise<void> {
-    await this._baseDatabase.removeEntities(dbTx, IPLDBlock, { relations: ['block'], where: { block: { blockNumber }, kind } });
+    const repo = dbTx.manager.getRepository(IPLDBlock);
+
+    await this._baseDatabase.removeIPLDBlocks(repo, blockNumber, kind);
   }
 
   async getHookStatus (): Promise<HookStatus | undefined> {

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

@@ -3,6 +3,9 @@
 //
 
 import { Entity, PrimaryGeneratedColumn, Column, Index, ManyToOne } from 'typeorm';
+
+import { StateKind } from '@vulcanize/util';
+
 import { BlockProgress } from './BlockProgress';
 
 @Entity()
@@ -22,8 +25,11 @@ export class IPLDBlock {
   @Column('varchar')
   cid!: string;
 
-  @Column('varchar')
-  kind!: string;
+  @Column({
+    type: 'enum',
+    enum: StateKind
+  })
+  kind!: StateKind;
 
   @Column('bytea')
   data!: Buffer;

packages/graph-test-watcher/src/hooks.ts

@@ -23,6 +23,7 @@ export async function createInitialState (indexer: Indexer, contractAddress: str
     state: {}
   };
 
+  // Return initial state data to be saved.
   return ipldBlockData;
 }
 
@@ -34,6 +35,8 @@ export async function createInitialState (indexer: Indexer, contractAddress: str
 export async function createStateDiff (indexer: Indexer, blockHash: string): Promise<void> {
   assert(indexer);
   assert(blockHash);
+
+  // Use indexer.createStateDiff() method to create a custom diff.
 }
 
 /**
@@ -48,6 +51,8 @@ export async function createStateCheckpoint (indexer: Indexer, contractAddress:
   assert(blockHash);
   assert(contractAddress);
 
+  // Use indexer.createStateCheckpoint() method to create a custom checkpoint.
+
   return false;
 }
 

packages/graph-test-watcher/src/indexer.ts

@@ -24,7 +24,8 @@ import {
   Where,
   QueryOptions,
   BlockHeight,
-  IPFSClient
+  IPFSClient,
+  StateKind
 } from '@vulcanize/util';
 import { GraphWatcher } from '@vulcanize/graph-node';
 
@@ -134,6 +135,7 @@ export class Indexer implements IndexerInterface {
 
   async init (): Promise<void> {
     await this._baseIndexer.fetchContracts();
+    await this._baseIndexer.fetchIPLDStatus();
   }
 
   getResultEvent (event: Event): ResultEvent {
@@ -254,6 +256,16 @@ export class Indexer implements IndexerInterface {
     await this._baseIndexer.pushToIPFS(data);
   }
 
+  async processInitialState (contractAddress: string, blockHash: string): Promise<any> {
+    // Call initial state hook.
+    return createInitialState(this, contractAddress, blockHash);
+  }
+
+  async processStateCheckpoint (contractAddress: string, blockHash: string): Promise<boolean> {
+    // Call checkpoint hook.
+    return createStateCheckpoint(this, contractAddress, blockHash);
+  }
+
   async processCanonicalBlock (job: any): Promise<void> {
     const { data: { blockHash } } = job;
 
@@ -281,7 +293,7 @@ export class Indexer implements IndexerInterface {
     return this._db.getPrevIPLDBlock(blockHash, contractAddress, kind);
   }
 
-  async getLatestIPLDBlock (contractAddress: string, kind: string | null, blockNumber?: number): Promise<IPLDBlock | undefined> {
+  async getLatestIPLDBlock (contractAddress: string, kind: StateKind | null, blockNumber?: number): Promise<IPLDBlock | undefined> {
     return this._db.getLatestIPLDBlock(contractAddress, kind, blockNumber);
   }
 
@@ -301,31 +313,40 @@ export class Indexer implements IndexerInterface {
     return this._baseIndexer.isIPFSConfigured();
   }
 
-  async createInitialState (contractAddress: string, blockHash: string): Promise<any> {
-    return createInitialState(this, contractAddress, blockHash);
-  }
-
+  // Method used to create auto diffs (diff_staged).
   async createDiffStaged (contractAddress: string, blockHash: string, data: any): Promise<void> {
     await this._baseIndexer.createDiffStaged(contractAddress, blockHash, data);
   }
 
+  // Method to be used by createStateDiff hook.
   async createDiff (contractAddress: string, blockHash: string, data: any): Promise<void> {
-    await this._baseIndexer.createDiff(contractAddress, blockHash, data);
+    const block = await this.getBlockProgress(blockHash);
+    assert(block);
+
+    await this._baseIndexer.createDiff(contractAddress, block, data);
   }
 
-  async createStateCheckpoint (contractAddress: string, blockHash: string): Promise<boolean> {
-    return createStateCheckpoint(this, contractAddress, blockHash);
+  // Method to be used by createStateCheckpoint hook.
+  async createStateCheckpoint (contractAddress: string, blockHash: string, data: any): Promise<void> {
+    const block = await this.getBlockProgress(blockHash);
+    assert(block);
+
+    return this._baseIndexer.createStateCheckpoint(contractAddress, block, data);
   }
 
-  async createCheckpoint (contractAddress: string, blockHash?: string, data?: any, checkpointInterval?: number): Promise<string | undefined> {
-    return this._baseIndexer.createCheckpoint(this, contractAddress, blockHash, data, checkpointInterval);
+  // Method to be used by checkpoint CLI.
+  async createCheckpoint (contractAddress: string, blockHash: string): Promise<string | undefined> {
+    const block = await this.getBlockProgress(blockHash);
+    assert(block);
+
+    return this._baseIndexer.createCheckpoint(this, contractAddress, block);
   }
 
   async saveOrUpdateIPLDBlock (ipldBlock: IPLDBlock): Promise<IPLDBlock> {
     return this._baseIndexer.saveOrUpdateIPLDBlock(ipldBlock);
   }
 
-  async removeIPLDBlocks (blockNumber: number, kind: string): Promise<void> {
+  async removeIPLDBlocks (blockNumber: number, kind: StateKind): Promise<void> {
     await this._baseIndexer.removeIPLDBlocks(blockNumber, kind);
   }
 
@@ -420,13 +441,12 @@ export class Indexer implements IndexerInterface {
   }
 
   async getLatestHooksProcessedBlock (): Promise<BlockProgress> {
-    const hookStatus = await this.getHookStatus();
-    assert(hookStatus);
-
-    return this._baseIndexer.getLatestHooksProcessedBlock(hookStatus);
+    return this._baseIndexer.getLatestHooksProcessedBlock();
   }
 
   async watchContract (address: string, kind: string, checkpoint: boolean, startingBlock: number): Promise<void> {
+    this._baseIndexer.updateIPLDStatusMap(address, {});
+
     return this._baseIndexer.watchContract(address, kind, checkpoint, startingBlock);
   }
 

packages/graph-test-watcher/src/job-runner.ts

@@ -78,11 +78,19 @@ export class JobRunner {
 
       const hookStatus = await this._indexer.getHookStatus();
 
-      if (hookStatus && hookStatus.latestProcessedBlockNumber < (blockNumber - 1)) {
-        const message = `Hooks for blockNumber ${blockNumber - 1} not processed yet, aborting`;
-        log(message);
+      if (hookStatus) {
+        if (hookStatus.latestProcessedBlockNumber < (blockNumber - 1)) {
+          const message = `Hooks for blockNumber ${blockNumber - 1} not processed yet, aborting`;
+          log(message);
 
-        throw new Error(message);
+          throw new Error(message);
+        }
+
+        if (hookStatus.latestProcessedBlockNumber > (blockNumber - 1)) {
+          log(`Hooks for blockNumber ${blockNumber} already processed`);
+
+          return;
+        }
       }
 
       await this._indexer.processCanonicalBlock(job);

packages/graph-test-watcher/src/resolvers.ts

@@ -8,7 +8,7 @@ import debug from 'debug';
 import Decimal from 'decimal.js';
 import { GraphQLScalarType } from 'graphql';
 
-import { ValueResult, BlockHeight, STATE_KIND_DIFF } from '@vulcanize/util';
+import { ValueResult, BlockHeight, StateKind } from '@vulcanize/util';
 
 import { Indexer } from './indexer';
 import { EventWatcher } from './events';
@@ -122,7 +122,7 @@ export const createResolvers = async (indexer: Indexer, eventWatcher: EventWatch
         return ipldBlock && ipldBlock.block.isComplete ? indexer.getResultIPLDBlock(ipldBlock) : undefined;
       },
 
-      getState: async (_: any, { blockHash, contractAddress, kind = STATE_KIND_DIFF }: { blockHash: string, contractAddress: string, kind: string }) => {
+      getState: async (_: any, { blockHash, contractAddress, kind = StateKind.Diff }: { blockHash: string, contractAddress: string, kind: string }) => {
         log('getState', blockHash, contractAddress, kind);
 
         const ipldBlock = await indexer.getPrevIPLDBlock(blockHash, contractAddress, kind);