Per Message TTL Caveats Documentation Update

[ramikhafagi96]

Is your feature request related to a problem? Please describe.

Problem Definition

In Time-To-Live and Expiration Caveats Section, It is mentioned that Only when expired messages reach the head of a queue will they actually be discarded (or dead-lettered). which is not true now. What actually happens is when the message TTL expires it gets discarded and dead-lettered even if it was not at the head of the queue. So, I think this was an old behavior given that it was written 6 years ago and it hasn't been updated since then.

Impact

Because of this out-dated documentation we took a design choice to implement delayed messages by creating many queues with different TTL on the queue itself and it made our design very complicated while it could've been much simpler if we used TTL on messages

Code Example

You can find below code example for showing that the mentioned behavior is not valid. The code example creates two queues the first one messages with expiration property are enqueued in it and when the TTL expires it gets dead-lettered to another queue and there's a running consumer that consumes from the dead-letter queue.

# publisher
import pika

# Connection parameters
url = "amqp://guest:guest@localhost:5672/%2F"
params = pika.URLParameters(url)

# Establish a connection with RabbitMQ server
connection = pika.BlockingConnection(params)
channel = connection.channel()

# Declare an exchange
exchange_name = "my-exchange"
channel.exchange_declare(exchange=exchange_name, exchange_type='direct')

# Declare queues
orders_queue = "orders-queue"
delay_queue = "delay-orders-queue"

channel.queue_declare(queue=orders_queue)
channel.queue_declare(queue=delay_queue, arguments={
    "x-dead-letter-exchange": exchange_name,
    "x-dead-letter-routing-key": "orders"
})

# Bind queues to the exchange
channel.queue_bind(queue=orders_queue, exchange=exchange_name, routing_key="orders")
channel.queue_bind(queue=delay_queue, exchange=exchange_name, routing_key="delayed-orders")

# Publish messages
message_body_no_ttl = 'delayed-message-no-ttl'
channel.basic_publish(exchange=exchange_name,
                      routing_key="delayed-orders",
                      body=message_body_no_ttl,
                      properties=pika.BasicProperties(content_type='text/plain'))

message_body_longer_ttl = "delayed-message-longer-ttl"
channel.basic_publish(exchange=exchange_name,
                      routing_key="delayed-orders",
                      body=message_body_longer_ttl,
                      properties=pika.BasicProperties(content_type='text/plain', expiration="10000"))

message_body_shorter_ttl = 'delayed-message-shorter-ttl'
channel.basic_publish(exchange=exchange_name,
                      routing_key="delayed-orders",
                      body=message_body_shorter_ttl,
                      properties=pika.BasicProperties(content_type='text/plain', expiration='5000'))

# Close the connection
connection.close()

# consumer
import pika

def on_message(channel, method_frame, header_frame, body):
    """Callback function to process the message."""
    print(f"Received message: {body.decode()}")
    # Acknowledge the message
    channel.basic_ack(delivery_tag=method_frame.delivery_tag)

# Connection parameters
url = "amqp://guest:guest@localhost:5672/%2F"
params = pika.URLParameters(url)

# Establish a connection with RabbitMQ server
connection = pika.BlockingConnection(params)
channel = connection.channel()

# Declare the queue (make sure it's the same name as the producer's queue)
queue_name = 'orders-queue'
channel.queue_declare(queue=queue_name)

# Start consuming messages from the queue
channel.basic_consume(queue=queue_name, on_message_callback=on_message, auto_ack=False)

print("Starting Consuming")
try:
    channel.start_consuming()
except KeyboardInterrupt:
    channel.stop_consuming()
except Exception as e:
    print(f"Error: {e}")
    channel.stop_consuming()

# Close the connection
connection.close()

Describe the solution you'd like

I suggest that this should be updated because it can lead to bad design choices

Describe alternatives you've considered

No response

Additional context

No response

[michaelklishin]

This is still just as true for many common cases. The rest of the section tries to get
the following point across:

When setting per-message TTL expired messages can queue up behind non-expired ones until the latter are consumed or expired

The current copywriting lumps together the cases when there is a DLX and when there isn't one (message TTL is often used as a way to control queue overflow, not for scheduling via DLX).

[ramikhafagi96]

But in my code example that's not the case, the queue already had an unconsumed message at the head and published messages with ttl when they got expired they were dead-lettered despite having a message with no expiration at the head of the queue. Doesn't that contradict with the documentation?

[michaelklishin]

@ramikhafagi96 I have updated my comment above. You are making an assumption that everyone uses TTL for scheduled dead lettering. That's not at all the case.

[michaelklishin]

There are several different scenarios that can have certain differences:

1. The queue is a classic queue, DLX is not used
2. The queue is a quorum queue, DLX is not used
3. The queue is a CQv2 and DLX is used
4. The queue is a CQv1 and DLX is used
5. The queue is a QQ and DLX is used

The current wording is correct for 1 and 2, you won't observe data deletion before messages reach the head.

Your example does not mention whether the classic queue used is a CQv2 queue or a CQv1 one. There may be subtle differences.

As for 5, expired messages won't be deleted until they reach the head of the queue but the dead lettering happens concurrently to that.

My guess is that when concurrent DLX was introduced to quorum queues, it was extended to CQs.

[ramikhafagi96]

@michaelklishin In my example I was using CQV2 as I was using rabbitmq v3.12.12.

[michaelklishin]

A quick search suggests that for QQs, the dead lettering process is indeed concurrent:

• A separate group of (Erlang) processes referred to internally as rabbit_fifo_dlx_worker performs the dead-lettering ("republishing") step
• Quorum queues have a separate internal queue of expired messages that the DLX worker uses as its input

However, that still does not mean that messages are marked for deletion immediately when they expire,
leave alone actually get removed instantly.

However, for classic queues things are much more involved and I am not convinced after a few minutes of digging in that your observations are universally true (they may be in your specific example, that I am not questioning).

Specifically, rabbit_amqqueue_process, the stateful part of a classic queue, has the following
comment and code snippet:

%% optimisation: it would be perfectly safe to always
%% invoke drop_expired_msgs here, but that is expensive so
%% we only do that if a new message that might have an
%% expiry ends up at the head of the queue. If the head
%% remains unchanged, or if the newly published message
%% has no expiry and becomes the head of the queue then
%% the call is unnecessary.
case {Dropped, QLen =:= 1, Props#message_properties.expiry} of
    {false, false,         _} -> State3;
    {true,  true,  undefined} -> State3;
    {_,     _,             _} -> drop_expired_msgs(State3)
end

In other words, classic queues in main and likely 3.12 will intentionally avoid dropping
(or dead lettering) messages in certain cases because doing so is not cheap and with some queue states, this would be counterproductive (e.g. if the queue <= 1 message).

Which means that for the sake of documentation, we can safely say that expired messages are
always dropped/dead lettered when they read the head of the queue. But I am not sure that we
can say that expired messages in the tail will be dead-lettered shortly after their expiration,
which is the case for QQs.

A single Python script won't provide an answer because you must have different load patterns
to observe different behaviors.

I will push some TTL guide doc updates that at least demonstrate the difference between the non-DLX and DLX cases (they simply should have separate doc sections), and that different queue types
actually drop/DLX messages at different moments because, well, different queue types have very different designs by definitions.

[ramikhafagi96]

Thank you so much for this thorough explanation. I should've tested stress tested that behavior in an environment close to production to verify that.

[michaelklishin]

@ramikhafagi96 while not directly related, note that your example

1. Does not use publisher confirms. They may have a practical effect on when queues perform certain activities. Needless to say we hope that most production systems do use publisher confirms
2. Closes the connection right after publishing. In other words, the channel is short lived. This does not affect queue operations directly but a closing channel will perform certain cleanup operations that may be subtly relevant

So while your findings is a good reminder that we need to re-organize the TTL guide, I am not convinced that every classic queue user will always observe the same behavior, for a number of unrelated and less-than-obvious reasons listed in this discussion. So we cannot make the docs say what you want them to say, not without a much more nuanced description after clarifying the current CQ behavior with the team members who have worked on CQv2 as well as message TTL support in QQs.

[michaelklishin]

a709816 introduces some initial edits that are badly needed.

The QQ and CQ sections can be clarified later. At the very least we no longer lump together like five different cases that really should be explained separately.

Thank you for bringing this up, @ramikhafagi96 👍

[michaelklishin]

After discussing it with two more team members, we will largely revert to the previous claim because QQs do not do concurrent dead lettering per se (that would make the FIFO guarantees highly dubious) but with just three messages in a queue and expiration in the range of [0; 10] seconds it is easy to arrive to the wrong conclusions.

The docs were not organized very well but they were factually correct.

[ramikhafagi96]

Thank you for the update 🙏🏻. I have a question, if I publish messages with per-message ttl to a queue and messages are published in the order of the expiration using publisher confirms, it would be safe to use that approach to have dynamic delay queue correct?