Skip to content

fix: Bug: robust_queue.bind on an expiring queue causes channel not found on ... - #3262

Closed
Roshan931 wants to merge 1 commit into
ag2ai:mainfrom
Roshan931:prilog/fix-error-rabbitdeclarer-caches-queues-the-broker-10-10-2026-97b28e
Closed

Roshan931 wants to merge 1 commit into
ag2ai:mainfrom
Roshan931:prilog/fix-error-rabbitdeclarer-caches-queues-the-broker-10-10-2026-97b28e

Conversation

@Roshan931

Copy link
Copy Markdown

Fixes #2721

Root cause

RabbitDeclarer caches queues the broker can delete (x-expires / auto_delete)

Changes

  • Bypass the declarer cache for queues RabbitMQ may delete behind our back (auto_delete, or an x-expires argument) so declare_queue re-sends the declaration and recreates the queue before it is bound; added regression tests asserting the second declare hits the channel.

Signed-off-by: Robert Markovski <5818108+Roshan931@users.noreply.github.com>
@CLAassistant

Copy link
Copy Markdown

CLA assistant check
Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
You have signed the CLA already but the status is still pending? Let us recheck it.

@IvanKirpichnikov

Copy link
Copy Markdown
Collaborator

Hi. I’m forced to close your Pull Request.

  1. I tried to run the MRE from the original issue.
    Of course, it didn’t run the first time. I modified it. It will be attached below. And the MRE ran without an error.

  2. I don’t like your solution. It seems like it should be implemented not at the FastStream level, but at the aio_pika level.
    Therefore, I ask you to look into the problem in more detail.

My MRE.

import asyncio

import aio_pika

from faststream.rabbit import ExchangeType, RabbitBroker
from faststream.rabbit.schemas import RabbitQueue, RabbitExchange
from faststream.rabbit.schemas.queue import ClassicQueueArgs


async def main():
    broker = RabbitBroker("amqp://guest:guest@localhost:5672/")

    exchange = RabbitExchange(
        name="amq.topic",
        type=ExchangeType.TOPIC,
        durable=True,
    )

    routing_key = "test.key"
    delay = 5  # seconds

    delay_queue = RabbitQueue(
        name=f"test.delay.{delay}",
        routing_key=f"{routing_key}.dlx.{delay}",
        durable=True,
        arguments=ClassicQueueArgs(
            {
                "x-dead-letter-exchange": exchange.name,
                "x-dead-letter-routing-key": routing_key,
                "x-message-ttl": delay * 1000,
                "x-expires": delay * 1000,
            }
        ),
    )

    final_queue = RabbitQueue(
        name="test.final",
        routing_key=routing_key,
        durable=True,
    )

    async with broker:
        await broker.declare_exchange(exchange)

        dq = await broker.declare_queue(delay_queue)
        fq = await broker.declare_queue(final_queue)

        await dq.bind(
            aio_pika.Exchange(
                dq.channel,
                name="amq.topic",
                type=ExchangeType.TOPIC,
                durable=True,
            ),
            routing_key=f"{routing_key}.dlx.{delay}",
        )
        await fq.bind(
            aio_pika.Exchange(
                fq.channel,
                name="amq.topic",
                type=ExchangeType.TOPIC,
                durable=True,
            ),
            routing_key=routing_key,
        )

        # publish message → should land in delay queue first,
        # then be dead-lettered to final queue after TTL
        await broker.publish(
            message={"hello": "world"},
            exchange=exchange.name,
            routing_key=f"{routing_key}.dlx.{delay}",
        )


if __name__ == "__main__":
    asyncio.run(main())

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Bug: robust_queue.bind on an expiring queue causes channel not found on same cases

3 participants