From 78b2f00ea99f9eecc36e4c7a93f7d4a575eb14c7 Mon Sep 17 00:00:00 2001
From: Scott Twiname <skott.twiname@gmail.com>
Date: Tue, 26 Nov 2024 15:08:01 +1300
Subject: [PATCH] Fix monitor write with workers, fix events iteration (#2607)

* Fix monitor write with workers, fix events iteration

* Update changelogs

* Fix optional value
---
 packages/node-core/CHANGELOG.md                    |  2 ++
 .../src/indexer/worker/worker.monitor.service.ts   | 14 ++++++++------
 packages/node/CHANGELOG.md                         |  2 ++
 packages/node/src/indexer/indexer.manager.ts       |  8 ++++----
 4 files changed, 16 insertions(+), 10 deletions(-)

diff --git a/packages/node-core/CHANGELOG.md b/packages/node-core/CHANGELOG.md
index 656ecf2a1e..6a33c8f2a2 100644
--- 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
index 62be0f64fb..9b766e92c4 100644
--- 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
index cbaf87e81c..6e2f3343f8 100644
--- 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
index dde11c5be7..bc19f5ffed 100644
--- 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);