Connecting Bullmq to AWS MemoryDB (tls, hash slot)

Ninthsun — Mon, 25 Sep 2023 12:37:33 +0000

While BullMQ is a powerful message queue which also can be implemented in Nestjs easily, it does not support tls connection or redis cluster very well.

I came across to small troubles when I was trying to use AWS MemoryDB with BullMQ. However, Using AWS MemoryDB instead of local redis server has two big difference.

1) uses tls
2) configured as a cluster

I'll go through these issues step by step. For those who aren't familiar with BullMQ in Nestjs, you can read BullMQ's Nestjs guide which is already pretty good.

TLS Connection

First big problem is that BullMQ does not provide a way to use tls connection, or a connection uri starting with rediss://.

BullModule.forRoot({
  connection: {
    host: 'localhost',
    port: 6379,
    tls: {
      ...
    },
  },
  prefix: 'bull',
})

host field only accepts the host part of the uri, and doesn't want to know about protocol or other query string options. You can still provide tls object, but you need certificates which is not quite necessary for connecting MemoryDB.

Luckily, connection field also accepts IORedis connection object, and IORedis connection object does provide a way to use connection string.

So, install ioredis and create a connection object.

BullModule.forRootAsync({
  imports: [ConfigModule],
  useFactory: (config: ConfigService) => {
    const uri = config.get<string>('redisUri');
    const connection = new IORedis(uri, {
      // you'll get some warnings if not set to null
      maxRetriesPerRequest: null,
    });
    return {
      connection,
      prefix: 'bull',
    };
  },
  inject: [ConfigService],
}),

Also, you can do the same with listeners.

async addQueue(data: YourData): Promise<JobResult> {
  const job = await this.queue.add(JOB_NAME, data, options);

  const listener = this.getListener();
  return job.waitUntilFinished(listener);
}

private getListener() {
  const uri = this.env.get<string>('redisUri');
  const connection = new IORedis(uri, { maxRetriesPerRequest: null });
  return new QueueEvents(QUEUE_NAME, { connection });
}

Hash Slots

Now we made a connection to AWS MemoryDB. But then, it is very likely to get an error saying,

Worker error ReplyError: CROSSSLOT Keys in request don't hash to the same slot

To understand what this error means, you first have to know how Redis Cluster stores many keys across its nodes. Redis Cluster implements a new identifier called slots. Each key is given a slot number, and this slot number is used as an index to find which key-value is in which node.

The problem is, redis transaction is only possible between the keys with same slots. Meaning you cannot make a transaction with key3and key4 from the image above, although they are in the same node. And this slot numbers are assigned by some hash algorithm based on the key. The resulted slots have maximum number up to 16384, still different keys are likely to have different slots assigned.

One way to control these assigned slots is to use hash tags. Hash tags is a part of key covered in brackets{}. For example, user1:session and user1:profile is not hash-tagged, so they are likely to get different hash slots. But, {user1}:session and {user1}:profile have hash tag string - user1, and this string is only hashed to get the slot. As a result, two keys are stored under same slot.

If you have a single queue within the same BullMQ connection, then hash tagging prefix like the BullMQ docs is good enough.

return {
  connection,
  prefix: '{bull}',
};

But if you use several queues with the same Redis connection, you can also hash tag queue names. In this case, I recommend creating a constant as queue name is quite frequently used across the code.

// const.ts
export const QueueConfig = {
  QUEUE_NAME: '{prompt}',
} as const;


// prompt.module.ts
@Module({
  imports: [
    BullModule.registerQueue({
      name: QueueConfig.QUEUE_NAME,
      defaultJobOptions: { ... },
    }),
  ],
  ...
})
export class PromptModule {}


// prompt.queue.ts
@Injectable()
@QueueEventsListener(QueueConfig.QUEUE_NAME)
export class PromptQueue extends QueueEventsHost {
  constructor(
    @InjectQueue(QueueConfig.QUEUE_NAME) private queue: Queue<PromptImageForm>,
  ) {
    super();
  }

  async add() { ... }
}

// prompt.processor.ts
@Processor(QueueConfig.QUEUE_NAME, { ... })
export class PromptProcessor extends WorkerHost { ... }

References

Jest worker encountered 4 child process exceptions, exceeding retry limit

Ninthsun — Sun, 25 Jun 2023 09:06:07 +0000

I ran across to this error while I was writing test code with Jest, being stunned by repeated long error messages in console ending with this test failure report.

While there might be several other reasons why this error might occur, in my case it was very simple - mishandled async error.

All you have to do is to check if you have any Promise not handled asynchronously.

Error reproduced



// user.controller.spec.ts
it('should throw Error if invalid user type.', () => {
  const result = controller.signup(signupForm, { type: 'invalid' });
  expect(result).rejects.toThrowError();
});

// user.controller.ts
async signup(form, query) {
  if (is<UserType>(query.type)) {
    throw new BadRequestException();
  }
  return true;
}

method signup throws an error asynchronously, and this is not handled in test code. For some reason, Jest considers test has gone wrong and retries it again in its other workers.

This error, can be suppressed if --maxWorker flag is set to 1 when running the test. However, this does not solve the problem from the ground, and one way to deal this issue is looking at async handling.

Solution

In the case above, only two words are added.

aysnc and await to where it should be.



it('should throw Error if invalid user type.', async () => {

  const result = controller.signup(signupForm, { type: 'invalid' });

  await expect(result).rejects.toThrowError();

});

P.S.

However, it's quite weird as this error does not always appear in similar situations. Seems like it is affected by other test cases.

DEV Community: Ninthsun