From b3821ad619cde66f36855c15405c4068332bde10 Mon Sep 17 00:00:00 2001 From: liuyi Date: Thu, 27 Feb 2025 14:25:46 +0800 Subject: [PATCH] fix(server): avoid global rejection when event handler errors (#10467) --- .../src/__tests__/event/eventbus.spec.ts | 54 ++++++++++++++++++- .../server/src/__tests__/event/provider.ts | 6 +++ .../backend/server/src/base/event/eventbus.ts | 29 ++++++++-- .../server/src/base/job/queue/index.ts | 12 ++++- .../server/src/plugins/payment/cron.ts | 26 ++++----- 5 files changed, 103 insertions(+), 24 deletions(-) diff --git a/packages/backend/server/src/__tests__/event/eventbus.spec.ts b/packages/backend/server/src/__tests__/event/eventbus.spec.ts index afb4b2305..4bf3b41b6 100644 --- a/packages/backend/server/src/__tests__/event/eventbus.spec.ts +++ b/packages/backend/server/src/__tests__/event/eventbus.spec.ts @@ -4,7 +4,7 @@ import { CLS_ID, ClsServiceManager } from 'nestjs-cls'; import Sinon from 'sinon'; import { EventBus, metrics } from '../../base'; -import { createTestingModule } from '../utils'; +import { createTestingModule, sleep } from '../utils'; import { Listeners } from './provider'; export const test = ava as TestFn<{ @@ -201,3 +201,55 @@ test('should continuously use the same request id', async t => { t.true(listeners.onRequestId.lastCall.returned('test-request-id')); }); + +test('should throw when emitting async event with uncaught error', async t => { + const { eventbus } = t.context; + + await t.throwsAsync( + () => eventbus.emitAsync('__test__.throw', { count: 0 }), + { + message: 'Error in event handler', + } + ); +}); + +test('should suppress thrown error when emitting async event', async t => { + const { eventbus } = t.context; + const spy = Sinon.spy(); + // @ts-expect-error internal event + const off = eventbus.on('error', spy); + + const promise = eventbus.emitAsync('__test__.suppressThrow', {}); + await t.notThrowsAsync(promise); + + t.true(spy.calledOnce); + const args = spy.firstCall.args[0]; + t.is(args.event, '__test__.suppressThrow'); + t.deepEqual(args.payload, {}); + t.is(args.error.message, 'Error in event handler'); + + const returns = await promise; + t.deepEqual(returns, [undefined]); + + off(); +}); + +test('should catch thrown error when emitting sync event', async t => { + const { eventbus } = t.context; + + const spy = Sinon.spy(); + // @ts-expect-error internal event + const off = eventbus.on('error', spy); + t.notThrows(() => eventbus.emit('__test__.throw', { count: 0 })); + + // wait a tick + await sleep(1); + + t.true(spy.calledOnce); + const args = spy.firstCall.args[0]; + t.is(args.event, '__test__.throw'); + t.deepEqual(args.payload, { count: 0 }); + t.is(args.error.message, 'Error in event handler'); + + off(); +}); diff --git a/packages/backend/server/src/__tests__/event/provider.ts b/packages/backend/server/src/__tests__/event/provider.ts index 58b67e9b6..1d707bdf5 100644 --- a/packages/backend/server/src/__tests__/event/provider.ts +++ b/packages/backend/server/src/__tests__/event/provider.ts @@ -8,6 +8,7 @@ declare global { '__test__.event': { count: number }; '__test__.event2': { count: number }; '__test__.throw': { count: number }; + '__test__.suppressThrow': {}; '__test__.requestId': {}; } } @@ -32,6 +33,11 @@ export class Listeners { throw new Error('Error in event handler'); } + @OnEvent('__test__.suppressThrow', { suppressError: true }) + onSuppressThrow() { + throw new Error('Error in event handler'); + } + @OnEvent('__test__.requestId') onRequestId() { const cls = ClsServiceManager.getClsService(); diff --git a/packages/backend/server/src/base/event/eventbus.ts b/packages/backend/server/src/base/event/eventbus.ts index f141ea3a1..a3ea945e4 100644 --- a/packages/backend/server/src/base/event/eventbus.ts +++ b/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(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; } diff --git a/packages/backend/server/src/base/job/queue/index.ts b/packages/backend/server/src/base/job/queue/index.ts index 2d3a124aa..3cde80a53 100644 --- a/packages/backend/server/src/base/job/queue/index.ts +++ b/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], diff --git a/packages/backend/server/src/plugins/payment/cron.ts b/packages/backend/server/src/plugins/payment/cron.ts index 0ef59dddf..d612a05af 100644 --- a/packages/backend/server/src/plugins/payment/cron.ts +++ b/packages/backend/server/src/plugins/payment/cron.ts @@ -2,7 +2,7 @@ import { Injectable } from '@nestjs/common'; import { Cron, CronExpression } from '@nestjs/schedule'; import { PrismaClient } from '@prisma/client'; -import { EventBus, JobQueue, OnEvent, OnJob } from '../../base'; +import { EventBus, JobQueue, OnJob } from '../../base'; import { SubscriptionPlan, SubscriptionRecurring, @@ -126,6 +126,15 @@ export class SubscriptionCronJobs { }); for (const subscription of subscriptions) { + await this.db.subscription.delete({ + where: { + targetId_plan: { + targetId: subscription.targetId, + plan: subscription.plan, + }, + }, + }); + this.event.emit('user.subscription.canceled', { userId: subscription.targetId, plan: subscription.plan as SubscriptionPlan, @@ -133,19 +142,4 @@ export class SubscriptionCronJobs { }); } } - - @OnEvent('user.subscription.canceled') - async handleUserSubscriptionCanceled({ - userId, - plan, - }: Events['user.subscription.canceled']) { - await this.db.subscription.delete({ - where: { - targetId_plan: { - targetId: userId, - plan, - }, - }, - }); - } }