Support filtering by topics when fetching logs in rpc-eth-client (#441)

* Support filtering by topics when fetching logs

* Include event signatures from all contracts in topics list

* Refactor common code

* Update package versions

packages/util/src/common.ts

@@ -258,9 +258,9 @@ export const _fetchBatchBlocks = async (
 export const processBatchEvents = async (indexer: IndexerInterface, block: BlockProgressInterface, eventsInBatch: number, subgraphEventsOrder: boolean): Promise<void> => {
   let dbBlock: BlockProgressInterface, dbEvents: EventInterface[];
   if (subgraphEventsOrder) {
-    ({ dbBlock, dbEvents } = await processEventsInSubgraphOrder(indexer, block, eventsInBatch));
+    ({ dbBlock, dbEvents } = await _processEventsInSubgraphOrder(indexer, block, eventsInBatch || DEFAULT_EVENTS_IN_BATCH));
   } else {
-    ({ dbBlock, dbEvents } = await processEvents(indexer, block, eventsInBatch));
+    ({ dbBlock, dbEvents } = await _processEvents(indexer, block, eventsInBatch || DEFAULT_EVENTS_IN_BATCH));
   }
 
   if (indexer.processBlockAfterEvents) {
@@ -279,25 +279,20 @@ export const processBatchEvents = async (indexer: IndexerInterface, block: Block
   console.timeEnd('time:common#processBatchEvents-updateBlockProgress-saveEvents');
 };
 
-export const processEvents = async (indexer: IndexerInterface, block: BlockProgressInterface, eventsInBatch: number): Promise<{ dbBlock: BlockProgressInterface, dbEvents: EventInterface[] }> => {
+const _processEvents = async (indexer: IndexerInterface, block: BlockProgressInterface, eventsInBatch: number): Promise<{ dbBlock: BlockProgressInterface, dbEvents: EventInterface[] }> => {
   const dbEvents: EventInterface[] = [];
-  let page = 0;
 
-  // Check if block processing is complete.
-  while (block.numProcessedEvents < block.numEvents) {
+  let page = 0;
+  let numFetchedEvents = 0;
+
+  // Check if we are out of events.
+  while (numFetchedEvents < block.numEvents) {
     console.time('time:common#processEvents-fetching_events_batch');
 
     // Fetch events in batches
-    const events = await indexer.getBlockEvents(
-      block.blockHash,
-      {},
-      {
-        skip: (page++) * (eventsInBatch || DEFAULT_EVENTS_IN_BATCH),
-        limit: eventsInBatch || DEFAULT_EVENTS_IN_BATCH,
-        orderBy: 'index',
-        orderDirection: OrderDirection.asc
-      }
-    );
+    const events = await _getEventsBatch(indexer, block.blockHash, eventsInBatch, page);
+    page++;
+    numFetchedEvents += events.length;
 
     console.timeEnd('time:common#processEvents-fetching_events_batch');
 
@@ -308,7 +303,7 @@ export const processEvents = async (indexer: IndexerInterface, block: BlockProgr
     console.time('time:common#processEvents-processing_events_batch');
 
     // Process events in loop
-    for (const event of events) {
+    for (let event of events) {
       // Skipping check for order of events processing since logIndex in FEVM is not index of log in block
       // Check was introduced to avoid reprocessing block events incase of restarts. But currently on restarts, unprocessed block is removed and reprocessed from first event log
       // if (event.index <= block.lastProcessedEventIndex) {
@@ -321,20 +316,8 @@ export const processEvents = async (indexer: IndexerInterface, block: BlockProgr
         // We might not have parsed this event yet. This can happen if the contract was added
         // as a result of a previous event in the same block.
         if (event.eventName === UNKNOWN_EVENT_NAME) {
-          const logObj = JSONbigNative.parse(event.extraInfo);
-
-          assert(indexer.parseEventNameAndArgs);
-          assert(typeof watchedContract !== 'boolean');
-          const { eventName, eventInfo, eventSignature } = indexer.parseEventNameAndArgs(watchedContract.kind, logObj);
-
-          event.eventName = eventName;
-          event.eventInfo = JSONbigNative.stringify(eventInfo);
-          event.extraInfo = JSONbigNative.stringify({
-            ...logObj,
-            eventSignature
-          });
-
-          // Save updated event to the db
+          // Parse the unknown event and save updated event to the db
+          event = _parseUnknownEvent(indexer, event, watchedContract.kind);
           dbEvents.push(event);
         }
 
@@ -351,30 +334,23 @@ export const processEvents = async (indexer: IndexerInterface, block: BlockProgr
   return { dbBlock: block, dbEvents };
 };
 
-export const processEventsInSubgraphOrder = async (indexer: IndexerInterface, block: BlockProgressInterface, eventsInBatch: number): Promise<{ dbBlock: BlockProgressInterface, dbEvents: EventInterface[] }> => {
+const _processEventsInSubgraphOrder = async (indexer: IndexerInterface, block: BlockProgressInterface, eventsInBatch: number): Promise<{ dbBlock: BlockProgressInterface, dbEvents: EventInterface[] }> => {
   // Create list of initially watched contracts
   const initiallyWatchedContracts: string[] = indexer.getWatchedContracts().map(contract => contract.address);
   const unwatchedContractEvents: EventInterface[] = [];
 
   const dbEvents: EventInterface[] = [];
+
   let page = 0;
+  let numFetchedEvents = 0;
 
   // Check if we are out of events.
-  let numFetchedEvents = 0;
   while (numFetchedEvents < block.numEvents) {
     console.time('time:common#processEventsInSubgraphOrder-fetching_events_batch');
 
     // Fetch events in batches
-    const events = await indexer.getBlockEvents(
-      block.blockHash,
-      {},
-      {
-        skip: (page++) * (eventsInBatch || DEFAULT_EVENTS_IN_BATCH),
-        limit: eventsInBatch || DEFAULT_EVENTS_IN_BATCH,
-        orderBy: 'index',
-        orderDirection: OrderDirection.asc
-      }
-    );
+    const events = await _getEventsBatch(indexer, block.blockHash, eventsInBatch, page);
+    page++;
     numFetchedEvents += events.length;
 
     console.timeEnd('time:common#processEventsInSubgraphOrder-fetching_events_batch');
@@ -397,12 +373,6 @@ export const processEventsInSubgraphOrder = async (indexer: IndexerInterface, bl
 
     // Process known events in a loop
     for (const event of watchedContractEvents) {
-      // Skipping check for order of events processing since logIndex in FEVM is not index of log in block
-      // Check was introduced to avoid reprocessing block events incase of restarts. But currently on restarts, unprocessed block is removed and reprocessed from first event log
-      // if (event.index <= block.lastProcessedEventIndex) {
-      //   throw new Error(`Events received out of order for block number ${block.blockNumber} hash ${block.blockHash}, got event index ${eventIndex} and lastProcessedEventIndex ${block.lastProcessedEventIndex}, aborting`);
-      // }
-
       await indexer.processEvent(event);
 
       block.lastProcessedEventIndex = event.index;
@@ -415,27 +385,15 @@ export const processEventsInSubgraphOrder = async (indexer: IndexerInterface, bl
   console.time('time:common#processEventsInSubgraphOrder-processing_unwatched_events');
 
   // At last, process all the events of newly watched contracts
-  for (const event of unwatchedContractEvents) {
+  for (let event of unwatchedContractEvents) {
     const watchedContract = indexer.isWatchedContract(event.contract);
 
     if (watchedContract) {
       // We might not have parsed this event yet. This can happen if the contract was added
       // as a result of a previous event in the same block.
       if (event.eventName === UNKNOWN_EVENT_NAME) {
-        const logObj = JSONbigNative.parse(event.extraInfo);
-
-        assert(indexer.parseEventNameAndArgs);
-        assert(typeof watchedContract !== 'boolean');
-        const { eventName, eventInfo, eventSignature } = indexer.parseEventNameAndArgs(watchedContract.kind, logObj);
-
-        event.eventName = eventName;
-        event.eventInfo = JSONbigNative.stringify(eventInfo);
-        event.extraInfo = JSONbigNative.stringify({
-          ...logObj,
-          eventSignature
-        });
-
-        // Save updated event to the db
+        // Parse the unknown event and save updated event to the db
+        event = _parseUnknownEvent(indexer, event, watchedContract.kind);
         dbEvents.push(event);
       }
 
@@ -451,6 +409,35 @@ export const processEventsInSubgraphOrder = async (indexer: IndexerInterface, bl
   return { dbBlock: block, dbEvents };
 };
 
+const _getEventsBatch = async (indexer: IndexerInterface, blockHash: string, eventsInBatch: number, page: number): Promise<EventInterface[]> => {
+  return indexer.getBlockEvents(
+    blockHash,
+    {},
+    {
+      skip: page * eventsInBatch,
+      limit: eventsInBatch,
+      orderBy: 'index',
+      orderDirection: OrderDirection.asc
+    }
+  );
+};
+
+const _parseUnknownEvent = (indexer: IndexerInterface, event: EventInterface, contractKind: string): EventInterface => {
+  const logObj = JSONbigNative.parse(event.extraInfo);
+
+  assert(indexer.parseEventNameAndArgs);
+  const { eventName, eventInfo, eventSignature } = indexer.parseEventNameAndArgs(contractKind, logObj);
+
+  event.eventName = eventName;
+  event.eventInfo = JSONbigNative.stringify(eventInfo);
+  event.extraInfo = JSONbigNative.stringify({
+    ...logObj,
+    eventSignature
+  });
+
+  return event;
+};
+
 /**
  * Create pruning job in QUEUE_BLOCK_PROCESSING.
  * @param jobQueue

packages/util/src/config.ts

@@ -200,7 +200,8 @@ export interface ServerConfig {
   subgraphPath: string;
   enableState: boolean;
   wasmRestartBlocksInterval: number;
-  filterLogs: boolean;
+  filterLogsByAddresses: boolean;
+  filterLogsByTopics: boolean;
   maxEventsBlockRange: number;
   clearEntitiesCacheInterval: number;
 

packages/util/src/indexer.ts

@@ -265,7 +265,7 @@ export class Indexer {
 
   // For each of the given blocks, fetches events and saves them along with the block to db
   // Returns an array with [block, events] for all the given blocks
-  async fetchEventsAndSaveBlocks (blocks: DeepPartial<BlockProgressInterface>[], parseEventNameAndArgs: (kind: string, logObj: any) => any): Promise<{ blockProgress: BlockProgressInterface, events: DeepPartial<EventInterface>[] }[]> {
+  async fetchEventsAndSaveBlocks (blocks: DeepPartial<BlockProgressInterface>[], eventSignaturesMap: Map<string, string[]>, parseEventNameAndArgs: (kind: string, logObj: any) => any): Promise<{ blockProgress: BlockProgressInterface, events: DeepPartial<EventInterface>[] }[]> {
     if (!blocks.length) {
       return [];
     }
@@ -274,7 +274,7 @@ export class Indexer {
     const toBlock = blocks[blocks.length - 1].blockNumber;
     log(`fetchEventsAndSaveBlocks#fetchEventsForBlocks: fetching from upstream server for range [${fromBlock}, ${toBlock}]`);
 
-    const dbEventsMap = await this.fetchEventsForBlocks(blocks, parseEventNameAndArgs);
+    const dbEventsMap = await this.fetchEventsForBlocks(blocks, eventSignaturesMap, parseEventNameAndArgs);
 
     const blocksWithEventsPromises = blocks.map(async block => {
       const blockHash = block.blockHash;
@@ -291,31 +291,25 @@ export class Indexer {
   }
 
   // Fetch events (to be saved to db) for a block range
-  async fetchEventsForBlocks (blocks: DeepPartial<BlockProgressInterface>[], parseEventNameAndArgs: (kind: string, logObj: any) => any): Promise<Map<string, DeepPartial<EventInterface>[]>> {
+  async fetchEventsForBlocks (blocks: DeepPartial<BlockProgressInterface>[], eventSignaturesMap: Map<string, string[]>, parseEventNameAndArgs: (kind: string, logObj: any) => any): Promise<Map<string, DeepPartial<EventInterface>[]>> {
     if (!blocks.length) {
       return new Map();
     }
 
     // Fetch logs for block range of given blocks
-    let logs: any;
     const fromBlock = blocks[0].blockNumber;
     const toBlock = blocks[blocks.length - 1].blockNumber;
 
     assert(this._ethClient.getLogsForBlockRange, 'getLogsForBlockRange() not implemented in ethClient');
-    if (this._serverConfig.filterLogs) {
-      const watchedContracts = this.getWatchedContracts();
-      const addresses = watchedContracts.map((watchedContract): string => {
-        return watchedContract.address;
-      });
 
-      ({ logs } = await this._ethClient.getLogsForBlockRange({
-        fromBlock,
-        toBlock,
-        addresses
-      }));
-    } else {
-      ({ logs } = await this._ethClient.getLogsForBlockRange({ fromBlock, toBlock }));
-    }
+    const { addresses, topics } = this._createLogsFilters(eventSignaturesMap);
+
+    const { logs } = await this._ethClient.getLogsForBlockRange({
+      fromBlock,
+      toBlock,
+      addresses,
+      topics
+    });
 
     // Skip further processing if no relevant logs found in the entire block range
     if (!logs.length) {
@@ -380,23 +374,15 @@ export class Indexer {
   }
 
   // Fetch events (to be saved to db) for a particular block
-  async fetchEvents (blockHash: string, blockNumber: number, parseEventNameAndArgs: (kind: string, logObj: any) => any): Promise<DeepPartial<EventInterface>[]> {
-    let logsPromise: Promise<any>;
+  async fetchEvents (blockHash: string, blockNumber: number, eventSignaturesMap: Map<string, string[]>, parseEventNameAndArgs: (kind: string, logObj: any) => any): Promise<DeepPartial<EventInterface>[]> {
+    const { addresses, topics } = this._createLogsFilters(eventSignaturesMap);
 
-    if (this._serverConfig.filterLogs) {
-      const watchedContracts = this.getWatchedContracts();
-      const addresses = watchedContracts.map((watchedContract): string => {
-        return watchedContract.address;
-      });
-
-      logsPromise = this._ethClient.getLogs({
-        blockHash,
-        blockNumber: blockNumber.toString(),
-        addresses
-      });
-    } else {
-      logsPromise = this._ethClient.getLogs({ blockHash, blockNumber: blockNumber.toString() });
-    }
+    const logsPromise = await this._ethClient.getLogs({
+      blockHash,
+      blockNumber: blockNumber.toString(),
+      addresses,
+      topics
+    });
 
     const transactionsPromise = this._ethClient.getBlockWithTransactions({ blockHash, blockNumber });
 
@@ -1198,6 +1184,29 @@ export class Indexer {
     this._stateStatusMap[address] = _.merge(oldStateStatus, stateStatus);
   }
 
+  _createLogsFilters (eventSignaturesMap: Map<string, string[]>): { addresses: string[] | undefined, topics: string[][] | undefined } {
+    let addresses: string[] | undefined;
+    let eventSignatures: string[] | undefined;
+
+    if (this._serverConfig.filterLogsByAddresses) {
+      const watchedContracts = this.getWatchedContracts();
+      addresses = watchedContracts.map((watchedContract): string => {
+        return watchedContract.address;
+      });
+    }
+
+    if (this._serverConfig.filterLogsByTopics) {
+      const eventSignaturesSet = new Set<string>();
+      eventSignaturesMap.forEach(sigs => sigs.forEach(sig => {
+        eventSignaturesSet.add(sig);
+      }));
+
+      eventSignatures = Array.from(eventSignaturesSet);
+    }
+
+    return { addresses, topics: eventSignatures && [eventSignatures] };
+  }
+
   parseEvent (logDescription: ethers.utils.LogDescription): { eventName: string, eventInfo: any, eventSignature: string } {
     const eventInfo = logDescription.eventFragment.inputs.reduce((acc: any, input, index) => {
       acc[input.name] = this._parseLogArg(input, logDescription.args[index]);

packages/util/src/types.ts

@@ -220,12 +220,14 @@ export interface EthClient {
   getLogs(vars: {
     blockHash: string,
     blockNumber: string,
-    addresses?: string[]
+    addresses?: string[],
+    topics?: string[][]
   }): Promise<any>;
   getLogsForBlockRange?: (vars: {
     fromBlock?: number,
     toBlock?: number,
-    addresses?: string[]
+    addresses?: string[],
+    topics?: string[][]
   }) => Promise<any>;
 }