Fix monitor write with workers, fix events iteration (#2607)

* Fix monitor write with workers, fix events iteration * Update changelogs * Fix optional value
subquery · Nov 26, 2024 · 78b2f00 · 78b2f00
1 parent 085765b
commit 78b2f00
Show file tree

Hide file tree

Showing 4 changed files with 16 additions and 10 deletions.
diff --git a/packages/node-core/CHANGELOG.md b/packages/node-core/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]
+### Fixed
+- Workers crashing because of lazy monitor write (#2607)
 
 ## [15.0.2] - 2024-11-26
 ### Fixed

diff --git a/packages/node-core/src/indexer/worker/worker.monitor.service.ts b/packages/node-core/src/indexer/worker/worker.monitor.service.ts
@@ -7,28 +7,30 @@ import {setMonitorService} from '../../process';
 import {MonitorServiceInterface} from '../monitor.service';
 
 type HostMonitorService = {
-  hostMonitorServiceWrite: (blockData: string) => void;
+  hostMonitorServiceWrite?: (blockData: string) => void;
 };
 
 export const hostMonitorKeys: (keyof HostMonitorService)[] = ['hostMonitorServiceWrite'];
 
-export function monitorHostFunctions(host: MonitorServiceInterface): HostMonitorService {
+export function monitorHostFunctions(host?: MonitorServiceInterface): HostMonitorService {
   return {
-    hostMonitorServiceWrite: host.write.bind(host),
+    hostMonitorServiceWrite: host?.write.bind(host),
   };
 }
 
 @Injectable()
 export class WorkerMonitorService implements MonitorServiceInterface {
-  constructor(private host: HostMonitorService) {
+  constructor(private host?: HostMonitorService) {
     if (isMainThread) {
       throw new Error('Expected to be worker thread');
     }
     setMonitorService(this);
   }
 
-  write(blockData: string): void {
-    return this.host?.hostMonitorServiceWrite ? this.host.hostMonitorServiceWrite(blockData) : undefined;
+  write(blockData: string | (() => string)): void {
+    return this.host?.hostMonitorServiceWrite
+      ? this.host.hostMonitorServiceWrite(typeof blockData === 'string' ? blockData : blockData())
+      : undefined;
   }
 
   createBlockFork(blockHeight: number): void {

diff --git a/packages/node/CHANGELOG.md b/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]
+### Fixed
+- Not using grouped events (#2607)
 
 ## [5.4.1] - 2024-11-26
 ### Fixed

diff --git a/packages/node/src/indexer/indexer.manager.ts b/packages/node/src/indexer/indexer.manager.ts
@@ -119,7 +119,7 @@ export class IndexerManager extends BaseIndexerManager<
             const idx = evt.extrinsic.idx;
             acc[idx] ??= [];
             acc[idx].push(evt);
-          } else {
+          } else if (!evt.phase.isApplyExtrinsic) {
             logger.warn(
               `Unrecognized event type, skipping. block="${block.block.header.number.toNumber()}" eventIdx="${idx}"`,
             );
@@ -141,9 +141,9 @@ export class IndexerManager extends BaseIndexerManager<
         await this.indexExtrinsic(extrinsic, dataSources, getVM);
 
         // Process extrinsic events
-        const extrinsicEvents = events
-          .filter((e) => e.extrinsic?.idx === extrinsic.idx)
-          .sort((a, b) => a.idx - b.idx);
+        const extrinsicEvents = (groupedEvents[extrinsic.idx] ?? []).sort(
+          (a, b) => a.idx - b.idx,
+        );
 
         for (const event of extrinsicEvents) {
           await this.indexEvent(event, dataSources, getVM);