Support/Mitigate RabbitMQ `consumer_timeout` behavior

[davidt99]

In RabbitMQ 3.8.15+, a new concept was introduced: consumer timeout. If a message is not acknowledged within the timeout period (defaulting to 30 minutes), RabbitMQ closes the connection and returns the message to the queue. This timeout can be configured globally, per-queue, or disabled (though disabling it is not recommended).

This behavior affects Dramatiq in several ways:

1. Reprocessing loops

If the TimeLimit middleware is not configured, or if its limit is higher than the RabbitMQ consumer timeout, a message that hits the timeout will be reprocessed indefinitely because Dramatiq loses the ability to ACK it once the connection drops.
A bug causing a worker to hang will eventually propagate to all workers as they all pick up the "poison" message, attempt to process it, and time out.

We should improve documentation regarding this. Additionally, I suggest adding a warning log when booting a RabbitMQ broker without the TimeLimit middleware. Adding TimeLimit by default seems incorrect as we cannot programmatically fetch the server's timeout value.

2. Delays exceeding the timeout

If a message has a delay (or backoff) longer than the consumer timeout, the Delayed Delivery (DQ) consumer thread will trigger a timeout. This causes all messages currently held in that thread to be returned to the queue. While this doesn't stop task processing entirely, it creates an unhealthy "churn" cycle.

I see here two changes we probably need to make:

1. Update DEFAULT_MAX_BACKOFF to 30 minutes when using the RabbitMQ broker to align with default server settings.
2. Modify the DQ consumer thread to re-enqueue messages at specific intervals (e.g., every 15 minutes) to avoid hitting the timeout. This is also a prerequisite for supporting Quorum Queues properly in the future.

3. Visibility and Logging

As discussed in #825, current logging can be misleading. In my experience, hunting down the specific message that exceeded 30 minutes was difficult because the exception is raised (and the connection closed) before the task actually finishes. We should improve the clarity of the logs to help users identify when a RabbitMQ-enforced timeout has occurred.

[LincolnPuzey]

Hi @davidt99 Thanks for the report.

For (1) I agree with your suggestions.
The good news is the default consumer-timeout is 30 minutes, and TimeLimit defaults to 10 minutes. So anyone using the defaults should be ok.

For (2). It would be good if we don't have to reduce DEFAULT_MAX_BACKOFF to 30 minutes. That would be a big change.
As you suggest, could we just re-queue messages so the timeout isn't reached?

For (3) Agreed if we can improve logging to make it more obvious what is happening, that would be good.

[davidt99]

@LincolnPuzey For (2), I think re-queue messages is the right approach as well, and it's something we will probably need to do when we support Quorum Queues because of the poison messages handling - When a messages is fetched and repeatedly re-queues, it will ended up in the dead-letter (after 20 times), this can happen when a worker is shutdown repeated times in an expected way (k8s scaling/deploying for example).
I tried to implement it, but the work_queue is a PriorityQueue sorted by the eta, which makes it a bit complex to iterate on the received time. I'll try to find an elegant way to implement this when I have time again.