x-death count not incremented when message expired

[bmflynn]

Describe the bug

When a message with an expiration and an existing x-death (same queue and reason) is re-published to a classic queue with a dead-letter exchange the x-death['count'] is not incremented.

Reproduction steps

import os
import time
from collections import namedtuple
from pprint import pprint

from pika import BasicProperties, BlockingConnection, URLParameters

Msg = namedtuple("Msg", ["method", "properties", "body"])

amqpurl = os.environ["AMQPURL"]
exchange = "amq.topic"

start_queue = "start_queue"
dlx_exchange = "errors"
error_queue = "dead_letters"
routing_key = "test_retries"

data = "XXX"

conn = BlockingConnection(URLParameters(amqpurl))
ch = conn.channel()

ch.exchange_declare(dlx_exchange, "topic")
ch.queue_declare(error_queue)
ch.queue_bind(error_queue, dlx_exchange, "#")

ch.queue_declare(start_queue, arguments={"x-dead-letter-exchange": dlx_exchange})
ch.queue_bind(start_queue, exchange, routing_key)

print(f"first publish {exchange=} {routing_key=} {data=}")
ch.basic_publish(exchange, routing_key, data, properties=BasicProperties(expiration="1000"))


# Wait for message to error out
print("waiting")
time.sleep(5)

first = Msg(*ch.basic_get(error_queue))
ch.basic_ack(first.method.delivery_tag)
assert first.method is not None, f"No message available {error_queue=}"
assert len(first.properties.headers["x-death"]) == 1
pprint(first.properties.headers)
assert first.properties.headers["x-death"][0]["count"] == 1

print("Second publish")
# Publish the message data again with the x-death headers having count=1
first.properties.expiration = "1000"
ch.basic_publish(exchange, routing_key, data, properties=first.properties)

# Wait for message to expire again
print("waiting")
time.sleep(5)
second = Msg(*ch.basic_get(error_queue))
ch.basic_ack(second.method.delivery_tag)
assert len(second.properties.headers["x-death"]) == 1
pprint(second.properties.headers)

ch.close()
conn.close()

Expected behavior

x-death['count'] should be incremented for each death with matching reason and queue per docs.

Additional context

I tested on versions 3.9, 3.10, 3.11, 3.12, & 3.13 using the rabbitmq:<ver>-management docker image and it only fails to increment on 3.13.

[kjnilsson]

did you enable the message_containers feature flag?

[bmflynn]

Not explicitly, but it does show as "Enabled" in the management interface.

[lukebakken]

@ansd and @kjnilsson discussed this internally and have found where the regression was introduced.

[bmflynn]

Great! I appreciate your time and effort.

…

On Sat, Mar 9, 2024, 10:20 AM Luke Bakken ***@***.***> wrote: @ansd <https://github.com/ansd> and @kjnilsson <https://github.com/kjnilsson> discussed this internally and have found where the regression was introduced. — Reply to this email directly, view it on GitHub <#10709 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ADDEIQQW2PECJJS6PLSD2PDYXMZF5AVCNFSM6AAAAABENLMJEGVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMYTSOBWHEYDGMBXHA> . You are receiving this because you authored the thread.Message ID: ***@***.***>

[kjnilsson]

@bmflynn could you possibly explain a bit how you are using this feature? Collaborative editing of x- headers is a bit odd in my view. x- headers "belong" to the broker and using it this way feels somewhat odd.

It is very unlikely we'd provide the same feature for other protocols such as AMQP (1.0)

[bmflynn]

Dead-lettering and the x-death['count'] value are used as part of a retry mechanism where a message with expiration is retried until x-death['count'] is greater than the maximum number of retries.

Essentially, when message handling fails the message properties (including headers) and data are published with a message expiration via an error exchange to a queue with the dead-letter exchange set to the original exchange. The message will remain in this queue to expire at which point it gets routed back to the original exchange and is retried. This mechanism continues until the original message is handled successfully or x-death['count'] reaches the configured maximum number of retries.

There is no editing of the x-headers other than the fact that they are part of the original message properties that are used as part of a new message that is sent to an error exchange.

What is the specific feature you are referring to? Do you mean it's unlikely that a death count, or the number of times a message is dead-lettered, will be available in AMQP 1.0?

[ansd]

@bmflynn Starting with "message containers" in 3.13, we would like to treat any x- headers as belonging to the broker, meaning a client shouldn't publish messages with x- headers. (If a client does publish with x- headers, these headers won't be interpreted in any special way anymore as illustrated by your example).

You are right that this is a behavioural change compared to 3.12 and arguably a breaking change belonging to 4.0.
So, we could "fix" / revert to the 3.12 behaviour by having the broker parse incoming headers and therefore interpreting the x-death header coming from a publisher.
However, we very likely would remove such special interpretation again in RabbitMQ 4.0 (which probably ships end of this year).
Therefore, my question is: Could your client set a custom non x- header and just increment it itself when it re-publishes the message?

Here is an example based on yours:

Code example with custom header

import os
import time
from collections import namedtuple
from pprint import pprint

from pika import BasicProperties, BlockingConnection, URLParameters

Msg = namedtuple("Msg", ["method", "properties", "body"])

amqpurl = "amqp://localhost:5672"
exchange = "amq.topic"

start_queue = "start_queue"
dlx_exchange = "errors"
error_queue = "dead_letters"
routing_key = "test_retries"

data = "XXX"

conn = BlockingConnection(URLParameters(amqpurl))
ch = conn.channel()

ch.exchange_declare(dlx_exchange, "topic")
ch.queue_declare(error_queue)
ch.queue_bind(error_queue, dlx_exchange, "#")

ch.queue_declare(start_queue, arguments={"x-dead-letter-exchange": dlx_exchange})
ch.queue_bind(start_queue, exchange, routing_key)

print(f"first publish {exchange=} {routing_key=} {data=}")
ch.basic_publish(exchange, routing_key, data, properties=BasicProperties(expiration="500"))

# Wait for message to error out
print("waiting\n")
time.sleep(1)

first = Msg(*ch.basic_get(error_queue))
ch.basic_ack(first.method.delivery_tag)
print("Received:")
pprint(first.properties.headers)

print("Second publish")
Key = "my-retry-count"
# Increment the retry count if it exists
first.properties.headers[Key] = first.properties.headers.get(Key, 0) + 1
first.properties.expiration = "500"
ch.basic_publish(exchange, routing_key, data, properties=first.properties)

# Wait for message to expire again
print("waiting\n")
time.sleep(1)
second = Msg(*ch.basic_get(error_queue))
ch.basic_ack(second.method.delivery_tag)
print("Received:")
pprint(second.properties.headers)

print("Third publish")
first.properties.headers[Key] = first.properties.headers.get(Key, 0) + 1
first.properties.expiration = "500"
ch.basic_publish(exchange, routing_key, data, properties=first.properties)

# Wait for message to expire again
print("waiting\n")
time.sleep(1)
second = Msg(*ch.basic_get(error_queue))
ch.basic_ack(second.method.delivery_tag)
print("Received:")
pprint(second.properties.headers)

ch.close()
conn.close()

[bmflynn]

@ansd Absolutely. I've made the change and our retries seem to be passing the same tests.

I think this issue is closable from my perspective. Thanks!