taskforcesh
[Bug]: FlowProducer throw Error event on Redis connection
Version
v5.31.2
Platform
NodeJS
What happened?
On my development environment, I randomly get this error after some time to use my NestJS dev server
node:events:496
api: throw er; // Unhandled 'error' event
api: ^
api: Error: read ECONNRESET
api: at TCP.onStreamRead (node:internal/stream_base_commons:218:20)
api: at TCP.callbackTrampoline (node:internal/async_hooks:130:17)
api: Emitted 'error' event on FlowProducer instance at:
api: at FlowProducer.emit (/Users/me/fabera/fabera-saas/node_modules/.pnpm/[email protected]/node_modules/bullmq/src/classes/flow-producer.ts:142:18)
api: at RedisConnection.<anonymous> (/Users/me/fabera/fabera-saas/node_modules/.pnpm/[email protected]/node_modules/bullmq/src/classes/flow-producer.ts:124:56)
api: at RedisConnection.emit (node:events:518:28)
api: at EventEmitter.RedisConnection.handleClientError (/Users/me/fabera/fabera-saas/node_modules/.pnpm/[email protected]/node_modules/bullmq/src/classes/redis-connection.ts:106:12)
api: at EventEmitter.emit (node:events:518:28)
api: at EventEmitter.silentEmit (/Users/me/fabera/fabera-saas/node_modules/.pnpm/[email protected]/node_modules/ioredis/built/Redis.js:482:30)
api: at Socket.<anonymous> (/Users/me/fabera/fabera-saas/node_modules/.pnpm/[email protected]/node_modules/ioredis/built/redis/event_handler.js:196:14)
api: at Object.onceWrapper (node:events:633:26)
api: at Socket.emit (node:events:530:35)
api: at emitErrorNT (node:internal/streams/destroy:169:8) {
api: errno: -54,
api: code: 'ECONNRESET',
api: syscall: 'read'
api: }
api: Node.js v20.18.1
If i relaunch the api, the error is solved. It could be a memory usage issue ? Or a bad dev server refresh handling ?
How to reproduce.
I have a queue and a flowProducer registered with the same name, but only the Flow producer throw this error
@Module({
imports: [
BullModule.registerFlowProducerAsync({
name: StepQueue.predict,
imports: [ConfigModule],
useFactory: async (configService: ConfigService) => ({
connection: {
host: configService.get('redisHost'),
port: configService.get('redisPort'),
username: configService.get('redisUsername'),
password: configService.get('redisPassword'),
},
defaultJobOptions: {
removeOnComplete: 200,
attempts: 5,
backoff: {
type: 'exponential',
delay: 1000,
},
},
}),
inject: [ConfigService],
}),
BullModule.registerQueueAsync({
name: StepQueue.predict,
imports: [ConfigModule],
useFactory: async (configService: ConfigService) => ({
connection: {
host: configService.get('redisHost'),
port: configService.get('redisPort'),
username: configService.get('redisUsername'),
password: configService.get('redisPassword'),
},
defaultJobOptions: {
removeOnComplete: 200,
attempts: 5,
backoff: {
type: 'exponential',
delay: 1000,
},
},
}),
inject: [ConfigService],
}),
]})
export class QueuesModule {}
Relevant log output
Code of Conduct
- [x] I agree to follow this project's Code of Conduct
Not sure exactly why you get this, but the error is a connection error to Redis, so I guess that the local redis server goes down or something. Btw, you should attach an error event handler to your flow producer instance to avoid unhandled exceptions.
Yes indeed, but why it happens to my flow producers and not the classic queues ? Btw my local redis database is always up