fix(server): avoid global rejection when event handler errors (#10467)

packages/backend/server/src/base/event/eventbus.ts

@@ -19,6 +19,12 @@ import { genRequestId } from '../utils';
 import { type EventName, type EventOptions } from './def';
 import { EventHandlerScanner } from './scanner';
 
+interface EventHandlerErrorPayload {
+  event: string;
+  payload: any;
+  error: Error;
+}
+
 /**
  * We use socket.io system to auto pub/sub on server to server broadcast events
  */
@@ -50,6 +56,9 @@ export class EventBus
 
   async onModuleInit() {
     this.bindEventHandlers();
+    this.emitter.on('error', ({ event, error }: EventHandlerErrorPayload) => {
+      this.logger.error(`Error happened when handling event ${event}`, error);
+    });
   }
 
   async onApplicationBootstrap() {
@@ -78,7 +87,16 @@ export class EventBus
    */
   emit<T extends EventName>(event: T, payload: Events[T]) {
     this.logger.log(`Dispatch event: ${event}`);
-    return this.emitter.emit(event, payload);
+
+    // NOTE(@forehalo):
+    //   Because all event handlers are wrapped in promisified metrics and cls context, they will always run in standalone tick.
+    //   In which way, if handler throws, an unhandled rejection will be triggered and end up with process exiting.
+    //   So we catch it here with `emitAsync`
+    this.emitter.emitAsync(event, payload).catch(e => {
+      this.emitter.emit('error', { event, payload, error: e });
+    });
+
+    return true;
   }
 
   /**
@@ -115,10 +133,11 @@ export class EventBus
             return await listener(payload);
           } catch (e) {
             if (suppressError) {
-              this.logger.error(
-                `Error happened when handling event ${signature}`,
-                e
-              );
+              this.emitter.emit('error', {
+                event,
+                payload,
+                error: e,
+              } as EventHandlerErrorPayload);
             } else {
               throw e;
             }

packages/backend/server/src/base/job/queue/index.ts

@@ -5,7 +5,7 @@ import { DynamicModule } from '@nestjs/common';
 
 import { Config } from '../../config';
 import { QueueRedis } from '../../redis';
-import { QUEUES } from './def';
+import { Queue, QUEUES } from './def';
 import { JobExecutor } from './executor';
 import { JobQueue } from './queue';
 import { JobHandlerScanner } from './scanner';
@@ -25,7 +25,15 @@ export class JobModule {
           },
           inject: [Config, QueueRedis],
         }),
-        BullModule.registerQueue(...QUEUES.map(name => ({ name }))),
+        BullModule.registerQueue(
+          ...QUEUES.map(name => {
+            if (name === Queue.NIGHTLY_JOB) {
+              // avoid nightly jobs been run multiple times
+              return { name, removeOnComplete: { age: 1000 * 60 * 60 } };
+            }
+            return { name };
+          })
+        ),
       ],
       providers: [JobQueue, JobExecutor, JobHandlerScanner],
       exports: [JobQueue],