Connecting Bullmq to AWS MemoryDB (tls, hash slot)
Tags: redis, aws, nestjs
Published: 2023-09-25 12:37:33

While [BullMQ](https://bullmq.io/) 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](https://docs.bullmq.io/guide/nestjs) 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://`.

```ts
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.

```ts
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.

```ts
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.

![Redis Cluster Hash Slots](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/mqc6hjiphiqrfv6db8sg.png)

The problem is, redis transaction is only possible between the keys with same slots. Meaning you cannot make a transaction with `key3`and `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](https://docs.bullmq.io/bull/patterns/redis-cluster) is good enough.

```ts
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.

```ts
// 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

- [[Resolved] CROSSSLOT Keys error in Redis cluster-mode enabled](https://inspector.dev/resolved-crossslot-keys-error-in-a-cluster-mode-enabled-redis/)
- [Redis - Redis Clustering Best Practices with Keys](https://redis.com/blog/redis-clustering-best-practices-with-keys/)