Redis Transport Based in Sorted Sets (BZMPOP)

[oliverhaas]

I've been playing around with changing how queues work in the Redis transport.

The main idea of this discussion: use sorted sets instead of lists.
Branch to check it out: [EDIT: Outdated] https://github.com/celery/kombu/tree/redis-transport-rework. The implementation is not yet cleaned up and could probably use a few pipelines or even lua scripts to make the whole thing more efficient.

Main points

• Queues are now sorted sets with BZMPOP instead of lists with BRPOP. This introduces the main downside: Requires Redis 7.0+ for BZMPOP, which I personally think is fine, but would mean that the old redis transport has to probably be supported for quite a while.
• Priority and timestamp are encoded in the score, so one sorted set per queue handles everything, which also removes the (soft) constraint that not too many priorities are supported, and the transport can easily support the 0-255 priorities like rabbitmq, which theoretically is also a breaking change compared to the old redis transport, but well, I think this makes more sense than to stay with the "old" implementation.
• Messages get stored in the hash at publish time, not consume time. The current implementation has tiny windows where messages can get lost. They're small, but they exist. No more "Well a message got lost, it could have been the transport.", which personally was a big problem for me and prompted this implementation.
• Negligibles downside: Officially O(log n) for insert/remove with sorted sets. But as a reference even for 1M entries - operations take ~100-200μs, which I think is fine for even a high volume message queue and quite a lot of messages in the queue. Also we do two fetches now to pull messages, since the queue and message content is stored separately, but I have at least one small idea to optimize this.
• visibility_timeout can now be MUCH smaller without any downside, since we're keeping track of the messages in another sorted set. Meaning even a visibility_timeout of like 3 minutes is feasible, and would mean a "stuck" message is directly detected much quicker and directly restored.

I might already start using this in production within the next month, but I would appreciate feedback here, especially if I'm missing something obvious.

[oliverhaas]

Here is a newer version of the idea as an actual transport in a package https://pypi.org/project/celery-redis-plus/. Sorted sets with 0-255 priority support, 100% reliability (current Kombu redis transport is NOT 100% reliable), less kinks related to visibility_timeout, Redis streams for fanout queues (also for reliability).

I will aim to submit this to Celery/Kombu somehow in the near future.