Should messages be manually deleted after consumption?

[hamez0r]

Hi, and thanks for this amazing library!

I'm using a POINT_TO_POINT queue and everything works great except for the fact that the number of Redis keys keeps growing, despite deliberately trying to not store messages.

This is my setup

const config: IRedisSMQConfig = {
  redis: {
    client: ERedisConfigClient.REDIS,
    options: {
      host: REDIS_HOST,
      port: REDIS_PORT,
      password: REDIS_PASS,
    },
  },
  messages: {
    store: {
      acknowledged: false,
      deadLettered: false,
    },
  },
};

Configuration.getSetConfig(config);

And this is my message handler

private async consume(msg: IMessageTransferable, done: ICallback<void>) {
  const messages = msg.body as Message[];

  try {
    await this.processMessages(messages);
  } catch (error) {
    console.error("Error processing messages:", error);
  }

  done(null);
}

Do I need to manually delete the messages after consumption?

Here are the keys that Redis has

127.0.0.1:6379> keys *
 1) "redis-smq-800.26:default:10:venue_6814f7cac1ab78ae0d8ea59a_messages"
 2) "redis-smq-800.26:global:19:d66ffef0-cfdc-46cb-b27a-c0c5160f25a8"
 3) "redis-smq-800.26:global:24:ef81967e-77f5-4e15-868b-44bfd1d46e3e"
 4) "redis-smq-800.26:global:24:3ac1bcfe-0a84-4f8a-b970-aec02dabd612"
 5) "redis-smq-800.26:global:24:3451c0cd-5719-4a97-8669-838fc03c53b4"
 6) "redis-smq-800.26:global:24:4f58dd25-410f-44de-b224-93238aea4cc2"
 7) "redis-smq-800.26:global:24:6230f4b8-a38f-435d-bfa9-539d7ad3c668"
 8) "redis-smq-800.26:global:24:84b3098f-7d0d-4658-9b38-cc82095cb430"
 9) "redis-smq-800.26:global:24:52723d21-6c9d-4187-a6ac-1506882afbb1"
10) "redis-smq-800.26:global:24:92287a72-7eae-4388-92dd-1f9c59962eb3"
11) "redis-smq-800.26:global:24:dbf57b6b-2588-471c-8016-895b84362407"
12) "redis-smq-800.26:global:24:547100a8-f662-436a-afeb-1c54698e4a19"
13) "redis-smq-800.26:global:24:1b52aec4-86bb-4e07-a853-b09c407e0786"
14) "redis-smq-800.26:global:24:f1695169-0bb3-4ad5-aad0-d55d217e3cfb"
15) "redis-smq-800.26:global:24:a9379c0a-6106-4c5e-8b19-c731caf61f54"
16) "redis-smq-800.26:global:24:18f079c6-e5f9-4fad-b775-dd5e31967966"
17) "redis-smq-800.26:global:24:5d2ae35c-0b10-4fa7-9172-c1f59bec456b"
18) "redis-smq-800.26:global:24:aea46694-eb55-4c80-8997-756ca99e757b"
19) "redis-smq-800.26:global:17"
20) "redis-smq-800.26:global:24:0b9df225-793a-4e0f-92ff-efa789f66598"
21) "redis-smq-800.26:global:24:e32ef133-48a9-457a-979d-c77bd859accf"
22) "redis-smq-800.26:global:24:b5cbd593-5c5c-4903-a307-d390f155107a"
23) "redis-smq-800.26:global:24:00709b81-92d1-4870-b4a1-ac062cd12423"
24) "redis-smq-800.26:global:18:d66ffef0-cfdc-46cb-b27a-c0c5160f25a8"
25) "redis-smq-800.26:global:24:76220ddf-9077-489b-ba60-1e87845ccd05"
26) "redis-smq-800.26:global:24:ae7d3bd7-0d6a-4e87-aaca-88df39e21a43"
27) "redis-smq-800.26:global:24:6b12fce9-4b41-4217-87a9-16dc06a4475c"
28) "redis-smq-800.26:global:24:95f98d87-9ea1-42c1-a6be-0ea416d23c13"
29) "redis-smq-800.26:global:24:53d8366f-76e3-45dd-a412-1ad6e44b0126"
30) "redis-smq-800.26:global:24:45a8782d-ddd3-491f-894f-a91bc7cba674"
31) "redis-smq-800.26:default:20"
32) "redis-smq-800.26:default:9:venue_6814f7cac1ab78ae0d8ea59a_messages"
33) "redis-smq-800.26:global:24:a602bfff-dd65-4b99-8a14-333e9a0e1519"
34) "redis-smq-800.26:global:24:f68d1604-c722-44b5-ab73-647e2e391b96"
35) "redis-smq-800.26:global:24:6182f1b7-0681-442f-8c38-ce86cb278c71"
36) "redis-smq-800.26:global:24:d068876b-fe74-4197-966a-1577ec1a04a4"
37) "redis-smq-800.26:default:14:venue_6814f7cac1ab78ae0d8ea59a_messages"
38) "redis-smq-800.26:global:24:6b9c0283-1569-4ce1-95fd-9eb965b2cc73"
39) "venue_6814f7cac1ab78ae0d8ea59a_event"
40) "redis-smq-800.26:global:24:0314f356-6169-4c3c-8223-74f17ec9aaf7"
41) "redis-smq-800.26:global:24:68ae1dd6-32ba-4c52-8a76-bc3f80ba78ad"
42) "redis-smq-800.26:global:24:04391042-ee07-41a2-a479-9f4d9a50b3c1"
43) "redis-smq-800.26:global:24:1afe375f-6883-4e87-ad36-da586c295d35"
44) "redis-smq-800.26:global:24:3a78df98-c518-4c25-ab69-22bd6ef7ae38"
45) "redis-smq-800.26:global:21"
46) "redis-smq-800.26:default:13:venue_6814f7cac1ab78ae0d8ea59a_messages"
47) "redis-smq-800.26:global:24:11360b0a-ffe1-477d-b349-b551d175ed94"
48) "redis-smq-800.26:default:11:venue_6814f7cac1ab78ae0d8ea59a_messages"
49) "redis-smq-800.26:global:24:de7aea71-88d6-4ac4-9293-9a6256fd0a05"
50) "redis-smq-800.26:global:24:a36131d2-2102-4ad7-9efe-0227bacca624"
51) "redis-smq-800.26:global:24:7ee0d13c-98be-4a99-8023-a431195fe45a"
52) "redis-smq-800.26:global:24:3ddba078-f1df-43ba-8ff5-0ce81674ae56"

[metsis]

This is exactly my issue as well. Messages in my queue are being produced and consumed without fail, but the 'global' keys keep growing as above.

[weyoss]

I believe that the documentation was not clear enough about message storage.

All published messages regardless of their statuses (scheduled, pending, acknowledged, dead-lettered, etc.) are stored in their respective queues until explicitly deleted.

RedisSMQ provide dedicated classes to manage acknowledged and dead-lettered messages. To make use of these classes dedicated storage for these types of messages needs to be enabled.

Please refer to https://github.com/weyoss/redis-smq/blob/master/packages/redis-smq/docs/message-storage.md for more details.

[hamez0r]

Hey! Thanks for the reply! That's fair enough. I do have a follow-up question, though: are messages in the acknowledged queue consumed by all consumers? Is it safe to delete them, once they're there?

[hamez0r]

For anyone getting this far in the thread, here's what I did. I save both acknowledged and dead lettered messages, as per documentation.

messages: {
  store: {
    acknowledged: true,
    deadLettered: true,
  },
},

The naive approach

I used an interval to instantiate a QueueAcknowledgedMessages and a QueueDeadLetteredMessages and .purge() everything, every 10 seconds.
Then something interesting happened: I started getting an error from Redis, saying the maximum client limit was reached. That was a bit baffling, since the default limit is 10k clients and I haven't changed it. After some investigation, I figured out that those 2 instances were leaving some dangling clients, and this is what was messing things up. Since then, I found out there's a .shutdown() method available on those classes, but I can't tell from the documentation if that actually closes the Redis clients.

The approach that worked

I instantiated the same classes (QueueAcknowledgedMessages and QueueDeadLetteredMessages) once for every queue, and I made them .purge() every 10 seconds. This solved the problem of having too many Redis clients.