Dead letter publisher patterns

[darioprencipe]

When handling an event with RabbitMQ broker using FastAPI plugin, I want to be able to send it to a dead letter exchange when an error occurs. While it's pretty clear from docs how I can declare a new exchange and parameterise it accordingly, I have some doubts on the right pattern to use with publishers.

Borrowing from Faststream docs page, if I understand correctly, I could do something like:

from faststream.exceptions import NackMessage
from faststream.rabbit import RabbitExchange, RabbitQueue
from faststream.rabbit.fastapi import (  # type: ignore
    Logger,
    RabbitMessage,
    RabbitRouter,
)


from internal.logic import compute_output
from internal.models import Input, Output

...

# Setup StreamRouter and instantiate RabbitMQ broker
router = RabbitRouter(
    settings.rabbitmq_url,
    schema_url="/asyncapi",
    include_in_schema=True,
)

broker = router.broker

# Config
exchange = RabbitExchange("my-exchange", durable=True, routing_key="") ### exchange
dlx = RabbitExchange("dead-letter-exchange", durable=True, routing_key="") ### dead letter exchange
queue_1 = RabbitQueue("queue-1", durable=True, routing_key="") ### the queue I'm listening to
queue_2 = RabbitQueue("queue-2", durable=True, routing_key="queue-2") ### the queue I'm publishing output into

# Instantiate publisher object
publisher = broker.publisher(queue=queue_2, exchange=exchange, persist=True)
# Instantiate dlx publisher object
dlx_publisher = broker.publisher(exchange=dlx, persist=True)


# Declare broker exchanges and queues on application startup
@router.after_startup
async def declare(app: FastAPI):
    await broker.declare_exchange(exchange)
    await broker.declare_exchange(dlx)
    await broker.declare_queue(queue_1)
    await broker.declare_queue(queue_2)


### Here I decorate my handler function with the publisher object that sends messages to my output queue
### BUT I also use dlx_publisher object declared above to push messages to dead letter exchange
@publisher
@router.subscriber(queue_1, exchange, retry=3)
async def handle_event(
    input: Input, msg: RabbitMessage, logger: Logger
) -> Output:
    try:
        output = compute_output(input)
        return output
    except Exception as err:
        logger.exception(err)
        dlx_publisher.publish(input)
        raise NackMessage()

# Create app
app = FastAPI(lifespan=router.lifespan_context)

# Include a StreamRouter object that uses RabbitMQ broker
app.include_router(router)

Do these (a) pattern and (b) config make sense?

On (a) pattern: not really clear what happens behind the scenes in Faststream event handling (@subscriber) when an exception is raised within the async function. Unless no_ack=True is specified (everything has to be handled manually)
On (b) config: is there any AioPika specific parameter to take into account when dealing with dead letter exchanges (and messages sent to them)?

Thanks in advance!

[Lancetnik]

@darioprencipe your code is working, but not using the native RMQ dead-letter features.
FastStream allows you to manual ack/nack message even without 'no_ack' argument (in this case it justdo nothing if message already acked inside your handler), but the preferred way - use RMQ dead-letter exchange and bind it to queue using queue bind arguments

https://www.rabbitmq.com/dlx.html#using-optional-queue-arguments

[darioprencipe]

@Lancetnik thanks a lot!

I refactored it as follows:

from fastapi import FastAPI, Request
from faststream.rabbit import ExchangeType, RabbitExchange, RabbitQueue
from faststream.rabbit.fastapi import (  # type: ignore
    Logger,
    RabbitMessage,
    RabbitRouter,
)

from internal.logic import compute_output
from internal.models import Input, Output

...

# Setup StreamRouter and instantiate RabbitMQ broker
router = RabbitRouter(
    settings.rabbitmq_url,
    schema_url="/asyncapi",
    include_in_schema=True,
)
broker = router.broker

# TODO: config
exchange = RabbitExchange(
    name="exchange",
    type=ExchangeType.DIRECT,
    durable=True,
)
dlx_exchange = RabbitExchange(name="dlx-exchange", durable=True)
queue_1 = RabbitQueue(
    name="queue-1",
    durable=True,
    arguments={
        "x-dead-letter-exchange": "dlx-exchange",
        "x-dead-letter-routing-key": "dlx-key",
        "x-queue-type": "quorum",
    },
)
queue_2 = RabbitQueue(
    name="queue-2",
    durable=True,
    arguments={
        "x-dead-letter-exchange": "dlx-exchange",
        "x-dead-letter-routing-key": "dlx-key",
        "x-queue-type": "quorum",
    },
)
dlq = RabbitQueue(name="dlq", durable=True)

# Instantiate publisher object
publisher = broker.publisher(queue=queue_2, exchange=exchange, persist=True)


# Declare broker exchanges and queues on application startup
@router.after_startup
async def declare(app: FastAPI):
    _exchange = await broker.declare_exchange(exchange)
    _dlx_exchange = await broker.declare_exchange(dlx_exchange)
    _queue_1 = await broker.declare_queue(queue_1)
    _queue_2 = await broker.declare_queue(queue_2)
    _dlq = await broker.declare_queue(dlq)
    await _queue_1.bind(_exchange)
    await _queue_2.bind(_exchange)
    await _dlq.bind(_dlx_exchange, routing_key="dlx-key")


# Define consumer and producer
### Decorating this function with @publisher and @subscriber implies:
### - given a message received on the subscribed queue, the (@publisher) decorated function output will be delivered to exchange
### - given a message received on the subscribed queue, the (@subscriber) decorated function takes message as input and execute function


@publisher
@router.subscriber(sync_queue, exchange, retry=3)
async def handle_event(input: Input, msg: RabbitMessage, logger: Logger) -> Output:
    try:
        output = compute_output(input)
        return output
    except Exception as err:
        logger.exception(err)
        raise NackMessage()



# Create app
app = FastAPI(lifespan=router.lifespan_context)

# Include a StreamRouter object that uses RabbitMQ broker
app.include_router(router)

What I don't understand from your answer is: once configured as above, do I need to explicitly nack (or reject) the message to push it to DLX? Or Faststream now will use RMQ features and will do it automatically when an exception occurs even if I don't make it explicit in my handler?

[Lancetnik]

@darioprencipe FastStream automatically acks message after successful processing or nacks it at fault (reject if retry limit exceeded), so you need no worry about

[darioprencipe]

@Lancetnik thanks a lot! 👏

[darioprencipe]

Hello @Lancetnik - could it be that another config / parameter might override the retry=N setup? Actually struggling with a retry=3 where a mock wrong event triggers consumers to crash forever (message is re-put int the queue over and over even if it should do so for 3 times max. Are you aware of any bugs/issues with that?

[Lancetnik]

@darioprencipe retry counter exists only in memory. It checks the message ID and calls RabbitMQ nack method at processing error or reject (with deletion msg from the queue) if message already nacked N (retry option) times. I don't know what exactly occurred in your case, but retry logic with in-memory counter can't be scaled by processes. Each instance has its own counter and message can be redelivered until any of this counter exceeded