Improve event iteration performance

packages/node/CHANGELOG.md

@@ -5,6 +5,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+### Changed
+- Improve event iteration, this improves performance with large blocks (#2601)
 
 ## [5.3.0] - 2024-11-22
 ### Added

packages/node/src/indexer/indexer.manager.ts

@@ -22,6 +22,7 @@ import {
   ProcessBlockResponse,
   BaseIndexerManager,
   IBlock,
+  getLogger,
 } from '@subql/node-core';
 import {
   LightSubstrateEvent,
@@ -39,6 +40,8 @@ import { DynamicDsService } from './dynamic-ds.service';
 import { ApiAt, BlockContent, isFullBlock, LightBlockContent } from './types';
 import { UnfinalizedBlocksService } from './unfinalizedBlocks.service';
 
+const logger = getLogger('indexer');
+
 @Injectable()
 export class IndexerManager extends BaseIndexerManager<
   ApiPromise,
@@ -105,9 +108,32 @@ export class IndexerManager extends BaseIndexerManager<
       const { block, events, extrinsics } = blockContent;
       await this.indexBlockContent(block, dataSources, getVM);
 
+      // Group the events so they only need to be iterated over a single time
+      const groupedEvents = events.reduce(
+        (acc, evt, idx) => {
+          if (evt.phase.isInitialization) {
+            acc.init.push(evt);
+          } else if (evt.phase.isFinalization) {
+            acc.finalize.push(evt);
+          } else if (evt.extrinsic?.idx) {
+            const idx = evt.extrinsic.idx;
+            acc[idx] ??= [];
+            acc[idx].push(evt);
+          } else {
+            logger.warn(
+              `Unrecognized event type, skipping. block="${block.block.header.number.toNumber()}" eventIdx="${idx}"`,
+            );
+          }
+          return acc;
+        },
+        { init: [], finalize: [] } as Record<
+          number | 'init' | 'finalize',
+          SubstrateEvent[]
+        >,
+      );
+
       // Run initialization events
-      const initEvents = events.filter((evt) => evt.phase.isInitialization);
-      for (const event of initEvents) {
+      for (const event of groupedEvents.init) {
         await this.indexEvent(event, dataSources, getVM);
       }
 
@@ -125,8 +151,7 @@ export class IndexerManager extends BaseIndexerManager<
       }
 
       // Run finalization events
-      const finalizeEvents = events.filter((evt) => evt.phase.isFinalization);
-      for (const event of finalizeEvents) {
+      for (const event of groupedEvents.finalize) {
         await this.indexEvent(event, dataSources, getVM);
       }
     } else {

packages/node/src/utils/substrate.ts

@@ -91,8 +91,9 @@ export function wrapExtrinsics(
   wrappedBlock: SubstrateBlock,
   allEvents: EventRecord[],
 ): SubstrateExtrinsic[] {
+  const groupedEvents = groupEventsByExtrinsic(allEvents);
   return wrappedBlock.block.extrinsics.map((extrinsic, idx) => {
-    const events = filterExtrinsicEvents(idx, allEvents);
+    const events = groupedEvents[idx];
     return {
       idx,
       extrinsic,
@@ -109,13 +110,22 @@ function getExtrinsicSuccess(events: EventRecord[]): boolean {
   );
 }
 
-function filterExtrinsicEvents(
-  extrinsicIdx: number,
+function groupEventsByExtrinsic(
   events: EventRecord[],
-): EventRecord[] {
-  return events.filter(
-    ({ phase }) =>
-      phase.isApplyExtrinsic && phase.asApplyExtrinsic.eqn(extrinsicIdx),
+): Record<number, EventRecord[]> {
+  return events.reduce(
+    (acc, event) => {
+      const extrinsicIdx = event.phase.isApplyExtrinsic
+        ? event.phase.asApplyExtrinsic.toNumber()
+        : undefined;
+      if (extrinsicIdx === undefined) {
+        return acc;
+      }
+      acc[extrinsicIdx] ??= [];
+      acc[extrinsicIdx].push(event);
+      return acc;
+    },
+    {} as Record<number, EventRecord[]>,
   );
 }
 
@@ -311,18 +321,6 @@ export async function getHeaderByHeight(
   return header;
 }
 
-export async function fetchBlocksRange(
-  api: ApiPromise,
-  startHeight: number,
-  endHeight: number,
-): Promise<SignedBlock[]> {
-  return Promise.all(
-    range(startHeight, endHeight + 1).map(async (height) =>
-      getBlockByHeight(api, height),
-    ),
-  );
-}
-
 export async function fetchBlocksArray(
   api: ApiPromise,
   blockArray: number[],