Changes to improve performance for event processing job (#304)

* Perf improvement TODOs

* Move watch contract to util package

* Get block progress from event instead of querying

* Use index field in event to check order of processing

* Use watched contract map to avoid querying database

* Use update query for blockProgress entity

Co-authored-by: Ashwin Phatak <ashwinpphatak@gmail.com>

packages/util/src/constants.ts

@@ -11,6 +11,9 @@ export const QUEUE_CHAIN_PRUNING = 'chain-pruning';
 export const JOB_KIND_INDEX = 'index';
 export const JOB_KIND_PRUNE = 'prune';
 
+export const JOB_KIND_CONTRACT = 'contract';
+export const JOB_KIND_EVENT = 'event';
+
 export const DEFAULT_CONFIG_PATH = 'environments/local.toml';
 
 export const UNKNOWN_EVENT_NAME = '__unknown__';

packages/util/src/database.ts

@@ -153,20 +153,20 @@ export class Database {
       .getMany();
   }
 
-  async updateBlockProgress (repo: Repository<BlockProgressInterface>, blockHash: string, lastProcessedEventIndex: number): Promise<void> {
-    const entity = await repo.findOne({ where: { blockHash } });
-    if (entity && !entity.isComplete) {
-      if (lastProcessedEventIndex <= entity.lastProcessedEventIndex) {
-        throw new Error(`Events processed out of order ${blockHash}, was ${entity.lastProcessedEventIndex}, got ${lastProcessedEventIndex}`);
+  async updateBlockProgress (repo: Repository<BlockProgressInterface>, block: BlockProgressInterface, lastProcessedEventIndex: number): Promise<void> {
+    if (!block.isComplete) {
+      if (lastProcessedEventIndex <= block.lastProcessedEventIndex) {
+        throw new Error(`Events processed out of order ${block.blockHash}, was ${block.lastProcessedEventIndex}, got ${lastProcessedEventIndex}`);
       }
 
-      entity.lastProcessedEventIndex = lastProcessedEventIndex;
-      entity.numProcessedEvents++;
-      if (entity.numProcessedEvents >= entity.numEvents) {
-        entity.isComplete = true;
+      block.lastProcessedEventIndex = lastProcessedEventIndex;
+      block.numProcessedEvents++;
+      if (block.numProcessedEvents >= block.numEvents) {
+        block.isComplete = true;
       }
 
-      await repo.save(entity);
+      const { id, ...blockData } = block;
+      await repo.update(id, blockData);
     }
   }
 
@@ -550,21 +550,24 @@ export class Database {
     return { canonicalBlockNumber, blockHashes };
   }
 
-  async getContract (repo: Repository<ContractInterface>, address: string): Promise<ContractInterface | undefined> {
+  async getContracts (repo: Repository<ContractInterface>): Promise<ContractInterface[]> {
     return repo.createQueryBuilder('contract')
-      .where('address = :address', { address })
-      .getOne();
+      .getMany();
   }
 
-  async saveContract (repo: Repository<ContractInterface>, address: string, startingBlock: number, kind?: string): Promise<void> {
-    const numRows = await repo
+  async saveContract (repo: Repository<ContractInterface>, address: string, startingBlock: number, kind?: string): Promise<ContractInterface> {
+    const contract = await repo
       .createQueryBuilder()
       .where('address = :address', { address })
-      .getCount();
+      .getOne();
 
-    if (numRows === 0) {
-      const entity = repo.create({ address, kind, startingBlock });
-      await repo.save(entity);
+    const entity = repo.create({ address, kind, startingBlock });
+
+    // If contract already present, overwrite fields.
+    if (contract) {
+      entity.id = contract.id;
     }
+
+    return repo.save(entity);
   }
 }

packages/util/src/indexer.ts

@@ -11,7 +11,8 @@ import { EthClient } from '@vulcanize/ipld-eth-client';
 import { GetStorageAt, getStorageValue, StorageLayout } from '@vulcanize/solidity-mapper';
 
 import { BlockProgressInterface, DatabaseInterface, EventInterface, SyncStatusInterface, ContractInterface } from './types';
-import { UNKNOWN_EVENT_NAME } from './constants';
+import { UNKNOWN_EVENT_NAME, JOB_KIND_CONTRACT, QUEUE_EVENT_PROCESSING } from './constants';
+import { JobQueue } from './job-queue';
 
 const MAX_EVENTS_BLOCK_RANGE = 1000;
 
@@ -30,15 +31,31 @@ export class Indexer {
   _postgraphileClient: EthClient;
   _getStorageAt: GetStorageAt;
   _ethProvider: ethers.providers.BaseProvider;
+  _jobQueue: JobQueue;
 
-  constructor (db: DatabaseInterface, ethClient: EthClient, postgraphileClient: EthClient, ethProvider: ethers.providers.BaseProvider) {
+  _watchedContracts: { [key: string]: ContractInterface } = {};
+
+  constructor (db: DatabaseInterface, ethClient: EthClient, postgraphileClient: EthClient, ethProvider: ethers.providers.BaseProvider, jobQueue: JobQueue) {
     this._db = db;
     this._ethClient = ethClient;
     this._postgraphileClient = postgraphileClient;
     this._ethProvider = ethProvider;
+    this._jobQueue = jobQueue;
     this._getStorageAt = this._ethClient.getStorageAt.bind(this._ethClient);
   }
 
+  async fetchContracts (): Promise<void> {
+    assert(this._db.getContracts);
+
+    const contracts = await this._db.getContracts();
+
+    this._watchedContracts = contracts.reduce((acc: { [key: string]: ContractInterface }, contract) => {
+      acc[contract.address] = contract;
+
+      return acc;
+    }, {});
+  }
+
   async getSyncStatus (): Promise<SyncStatusInterface | undefined> {
     const dbTx = await this._db.createTransactionRunner();
     let res;
@@ -152,12 +169,12 @@ export class Indexer {
     }
   }
 
-  async updateBlockProgress (blockHash: string, lastProcessedEventIndex: number): Promise<void> {
+  async updateBlockProgress (block: BlockProgressInterface, lastProcessedEventIndex: number): Promise<void> {
     const dbTx = await this._db.createTransactionRunner();
     let res;
 
     try {
-      res = await this._db.updateBlockProgress(dbTx, blockHash, lastProcessedEventIndex);
+      res = await this._db.updateBlockProgress(dbTx, block, lastProcessedEventIndex);
       await dbTx.commitTransaction();
     } catch (error) {
       await dbTx.rollbackTransaction();
@@ -281,9 +298,39 @@ export class Indexer {
   }
 
   async isWatchedContract (address : string): Promise<ContractInterface | undefined> {
-    assert(this._db.getContract);
+    return this._watchedContracts[address];
+  }
 
-    return this._db.getContract(ethers.utils.getAddress(address));
+  async watchContract (address: string, kind: string, startingBlock: number): Promise<void> {
+    assert(this._db.saveContract);
+    const dbTx = await this._db.createTransactionRunner();
+
+    // Always use the checksum address (https://docs.ethers.io/v5/api/utils/address/#utils-getAddress).
+    const contractAddress = ethers.utils.getAddress(address);
+
+    try {
+      const contract = await this._db.saveContract(dbTx, contractAddress, kind, startingBlock);
+      this.cacheContract(contract);
+      await dbTx.commitTransaction();
+
+      await this._jobQueue.pushJob(
+        QUEUE_EVENT_PROCESSING,
+        {
+          kind: JOB_KIND_CONTRACT,
+          contract
+        },
+        { priority: 1 }
+      );
+    } catch (error) {
+      await dbTx.rollbackTransaction();
+      throw error;
+    } finally {
+      await dbTx.release();
+    }
+  }
+
+  cacheContract (contract: ContractInterface): void {
+    this._watchedContracts[contract.address] = contract;
   }
 
   async getStorageValue (storageLayout: StorageLayout, blockHash: string, token: string, variable: string, ...mappingKeys: any[]): Promise<ValueResult> {

packages/util/src/job-queue.ts

@@ -52,6 +52,10 @@ export class JobQueue {
     await this._boss.start();
   }
 
+  async stop (): Promise<void> {
+    await this._boss.stop();
+  }
+
   async subscribe (queue: string, callback: JobCallback): Promise<string> {
     return await this._boss.subscribe(
       queue,

packages/util/src/job-runner.ts

@@ -8,7 +8,7 @@ import { wait } from './misc';
 import { createPruningJob } from './common';
 
 import { JobQueueConfig } from './config';
-import { JOB_KIND_INDEX, JOB_KIND_PRUNE, MAX_REORG_DEPTH, QUEUE_BLOCK_PROCESSING, QUEUE_EVENT_PROCESSING } from './constants';
+import { JOB_KIND_INDEX, JOB_KIND_PRUNE, JOB_KIND_EVENT, JOB_KIND_CONTRACT, MAX_REORG_DEPTH, QUEUE_BLOCK_PROCESSING, QUEUE_EVENT_PROCESSING } from './constants';
 import { JobQueue } from './job-queue';
 import { EventInterface, IndexerInterface, SyncStatusInterface } from './types';
 
@@ -46,34 +46,20 @@ export class JobRunner {
     }
   }
 
-  async processEvent (job: any): Promise<EventInterface> {
-    const { data: { id } } = job;
+  async processEvent (job: any): Promise<EventInterface | void> {
+    const { data: { kind } } = job;
 
-    log(`Processing event ${id}`);
+    switch (kind) {
+      case JOB_KIND_EVENT:
+        return this._processEvent(job);
 
-    const dbEvent = await this._indexer.getEvent(id);
-    assert(dbEvent);
+      case JOB_KIND_CONTRACT:
+        return this._updateWatchedContracts(job);
 
-    const event = dbEvent;
-
-    const blockProgress = await this._indexer.getBlockProgress(event.block.blockHash);
-    assert(blockProgress);
-
-    const events = await this._indexer.getBlockEvents(event.block.blockHash);
-    const eventIndex = events.findIndex((e: any) => e.id === event.id);
-    assert(eventIndex !== -1);
-
-    // Check if previous event in block has been processed exactly before this and abort if not.
-    if (eventIndex > 0) { // Skip the first event in the block.
-      const prevIndex = eventIndex - 1;
-      const prevEvent = events[prevIndex];
-      if (prevEvent.index !== blockProgress.lastProcessedEventIndex) {
-        throw new Error(`Events received out of order for block number ${event.block.blockNumber} hash ${event.block.blockHash},` +
-        ` prev event index ${prevEvent.index}, got event index ${event.index} and lastProcessedEventIndex ${blockProgress.lastProcessedEventIndex}, aborting`);
-      }
+      default:
+        log(`Invalid Job kind ${kind} in QUEUE_EVENT_PROCESSING.`);
+        break;
     }
-
-    return event;
   }
 
   async _pruneChain (job: any, syncStatus: SyncStatusInterface): Promise<void> {
@@ -180,8 +166,37 @@ export class JobRunner {
       const events = await this._indexer.getOrFetchBlockEvents({ blockHash, blockNumber, parentHash, blockTimestamp: timestamp });
 
       for (let ei = 0; ei < events.length; ei++) {
-        await this._jobQueue.pushJob(QUEUE_EVENT_PROCESSING, { id: events[ei].id, publish: true });
+        await this._jobQueue.pushJob(QUEUE_EVENT_PROCESSING, { kind: JOB_KIND_EVENT, id: events[ei].id, publish: true });
       }
     }
   }
+
+  async _processEvent (job: any): Promise<EventInterface> {
+    const { data: { id } } = job;
+
+    log(`Processing event ${id}`);
+
+    const event = await this._indexer.getEvent(id);
+    assert(event);
+    const eventIndex = event.index;
+
+    // Check if previous event in block has been processed exactly before this and abort if not.
+    if (eventIndex > 0) { // Skip the first event in the block.
+      const prevIndex = eventIndex - 1;
+
+      if (prevIndex !== event.block.lastProcessedEventIndex) {
+        throw new Error(`Events received out of order for block number ${event.block.blockNumber} hash ${event.block.blockHash},` +
+        ` prev event index ${prevIndex}, got event index ${event.index} and lastProcessedEventIndex ${event.block.lastProcessedEventIndex}, aborting`);
+      }
+    }
+
+    return event;
+  }
+
+  async _updateWatchedContracts (job: any): Promise<void> {
+    const { data: { contract } } = job;
+
+    assert(this._indexer.cacheContract);
+    this._indexer.cacheContract(contract);
+  }
 }

packages/util/src/types.ts

@@ -56,11 +56,12 @@ export interface IndexerInterface {
   getAncestorAtDepth (blockHash: string, depth: number): Promise<string>
   getOrFetchBlockEvents (block: DeepPartial<BlockProgressInterface>): Promise<Array<EventInterface>>
   removeUnknownEvents (block: BlockProgressInterface): Promise<void>
-  updateBlockProgress (blockHash: string, lastProcessedEventIndex: number): Promise<void>
+  updateBlockProgress (block: BlockProgressInterface, lastProcessedEventIndex: number): Promise<void>
   updateSyncStatusChainHead (blockHash: string, blockNumber: number): Promise<SyncStatusInterface>
   updateSyncStatusIndexedBlock (blockHash: string, blockNumber: number, force?: boolean): Promise<SyncStatusInterface>
   updateSyncStatusCanonicalBlock (blockHash: string, blockNumber: number, force?: boolean): Promise<SyncStatusInterface>
   markBlocksAsPruned (blocks: BlockProgressInterface[]): Promise<void>;
+  cacheContract?: (contract: ContractInterface) => void;
 }
 
 export interface EventWatcherInterface {
@@ -80,12 +81,13 @@ export interface DatabaseInterface {
   getProcessedBlockCountForRange (fromBlockNumber: number, toBlockNumber: number): Promise<{ expected: number, actual: number }>;
   getEventsInRange (fromBlockNumber: number, toBlockNumber: number): Promise<Array<EventInterface>>;
   markBlocksAsPruned (queryRunner: QueryRunner, blocks: BlockProgressInterface[]): Promise<void>;
-  updateBlockProgress (queryRunner: QueryRunner, blockHash: string, lastProcessedEventIndex: number): Promise<void>
+  updateBlockProgress (queryRunner: QueryRunner, block: BlockProgressInterface, lastProcessedEventIndex: number): Promise<void>
   updateSyncStatusIndexedBlock (queryRunner: QueryRunner, blockHash: string, blockNumber: number, force?: boolean): Promise<SyncStatusInterface>;
   updateSyncStatusChainHead (queryRunner: QueryRunner, blockHash: string, blockNumber: number): Promise<SyncStatusInterface>;
   updateSyncStatusCanonicalBlock (queryRunner: QueryRunner, blockHash: string, blockNumber: number, force?: boolean): Promise<SyncStatusInterface>;
   saveEvents (queryRunner: QueryRunner, block: DeepPartial<BlockProgressInterface>, events: DeepPartial<EventInterface>[]): Promise<void>;
   saveEventEntity (queryRunner: QueryRunner, entity: EventInterface): Promise<EventInterface>;
   removeEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, findConditions?: FindManyOptions<Entity> | FindConditions<Entity>): Promise<void>;
-  getContract?: (address: string) => Promise<ContractInterface | undefined>
+  getContracts?: () => Promise<ContractInterface[]>
+  saveContract?: (queryRunner: QueryRunner, contractAddress: string, kind: string, startingBlock: number) => Promise<ContractInterface>
 }