Closing connections with RabbitMQ leads to warnings in the broker side

[stronk7]

Describe the bug While advancing with some application using faststream, by coincidence I went to look to RabbitMQ logs and found lots of warnings about unexpected closed connections.

After debugging a lot locally, initially expecting that it was some bug my side, I think that I'm doing things properly (start & close). And have reduced the problem to a minimal example.

Note I'm a noob and maybe I'm missing something, but haven't been able to find anything in docs, issues, discussions about how to, properly, stop/close a connection. Also note that this problem, apparently, doesn't cause any trouble and everything seems to be working ok, just it's annoying to get all those warnings in RabbitMQ logs.

How to reproduce

import asyncio

from faststream.rabbit import RabbitBroker

# Usage example
async def main():

    broker = RabbitBroker("amqp://user:pass@localhost:5672")

    await broker.start()
    await asyncio.sleep(3)
    await broker.close()

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

And/Or steps to reproduce the behavior:

1. Run the script.

Expected behavior Broker disconnects properly and no problems are logged in the Rabbit side.

Observed behavior In Rabbit logs, the execution of the script leads to these logs:

2024-12-15 19:17:05.933765+00:00 [info] <0.10969.0> accepting AMQP connection 172.17.0.1:65170 -> 172.17.0.3:5672
2024-12-15 19:17:05.942399+00:00 [info] <0.10969.0> connection 172.17.0.1:65170 -> 172.17.0.3:5672: user 'user' authenticated and granted access to vhost '/'
2024-12-15 19:17:08.979577+00:00 [warning] <0.10969.0> closing AMQP connection <0.10969.0> (172.17.0.1:65170 -> 172.17.0.3:5672, vhost: '/', user: 'user', duration: '3s'):
2024-12-15 19:17:08.979577+00:00 [warning] <0.10969.0> client unexpectedly closed TCP connection

Screenshots N/A

Environment

$ faststream -v
Running FastStream 0.5.33 with CPython 3.12.7 on Darwin

Additional context RabbitMQ running locally, docker.

[stronk7]

In the other side, using a "sister" aio-pika (9.5.3) implementation, it seems to work.

import asyncio

import aio_pika
from yarl import URL


# Usage example
async def main():

    broker = aio_pika.connection.Connection(URL("amqp://user:pass@localhost:5672"))

    await broker.connect()
    await asyncio.sleep(3)
    await broker.close()

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

And logs look ok:

2024-12-15 22:28:31.864364+00:00 [info] <0.11776.0> accepting AMQP connection 172.17.0.1:58520 -> 172.17.0.3:5672
2024-12-15 22:28:31.868350+00:00 [info] <0.11776.0> connection 172.17.0.1:58520 -> 172.17.0.3:5672: user 'user' authenticated and granted access to vhost '/'
2024-12-15 22:28:34.870976+00:00 [info] <0.11776.0> closing AMQP connection (172.17.0.1:58520 -> 172.17.0.3:5672, vhost: '/', user: 'user', duration: '3s')

^^^ Also, I've tried with aio_pika.connect_robust() all ok too.

[Lancetnik]

Sorry, but I can't reproduce the problem on my side - your example works as expected with our docker-compose 🤔

import asyncio

from faststream.rabbit import RabbitBroker

async def main() -> None:
    broker = RabbitBroker()

    await broker.start()
    await asyncio.sleep(3.0)
    await broker.close()

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

2024-12-16 07:35:34.316158+00:00 [info] <0.712.0> accepting AMQP connection <0.712.0> (192.168.65.1:30176 -> 172.19.0.5:5672)
2024-12-16 07:35:34.322551+00:00 [info] <0.712.0> connection <0.712.0> (192.168.65.1:30176 -> 172.19.0.5:5672): user 'guest' authenticated and granted access to vhost '/'
2024-12-16 07:35:37.338315+00:00 [info] <0.712.0> closing AMQP connection <0.712.0> (192.168.65.1:30176 -> 172.19.0.5:5672, vhost: '/', user: 'guest')

Seems like it is an environment problem and we should take more time to dig into

Thank your for the Issue and for checking aio-pika case!

[NelsonNotes]

@Lancetnik assign

[NelsonNotes]

I did a little research and reproduced this behavior by breaking the FastStream rabbit connection close method. Here is code snippet:

class RabbitBroker(
    # ...
):
    # ...
    async def _close(
        self,
        exc_type: Optional[Type[BaseException]] = None,
        exc_val: Optional[BaseException] = None,
        exc_tb: Optional["TracebackType"] = None,
    ) -> None:
        if self._channel is not None:
            # if not self._channel.is_closed:  # channel closing was commented out
            #     await self._channel.close()

            self._channel = None

        self.declarer = None
        self._producer = None

        if self._connection is not None:
            await self._connection.close()

        await super()._close(exc_type, exc_val, exc_tb)
    # ...

After these changes I started getting the same warnings from RabbitMQ:

2025-03-17 08:55:29 2025-03-17 05:55:29.229818+00:00 [info] <0.2382.0> accepting AMQP connection 172.23.0.1:58284 -> 172.23.0.2:5672
2025-03-17 08:55:29 2025-03-17 05:55:29.233339+00:00 [info] <0.2382.0> connection 172.23.0.1:58284 -> 172.23.0.2:5672: user 'guest' authenticated and granted access to vhost '/'
2025-03-17 08:55:32 2025-03-17 05:55:32.133272+00:00 [warning] <0.2382.0> closing AMQP connection <0.2382.0> (172.23.0.1:58284 -> 172.23.0.2:5672, vhost: '/', user: 'guest', duration: '2s'):
2025-03-17 08:55:32 2025-03-17 05:55:32.133272+00:00 [warning] <0.2382.0> client unexpectedly closed TCP connection

I don't know if I'm on the right track, but at least I was able to reproduce it. I guess that aio-pika channel is not being closed properly in some cases.

@stronk7 can you please share your RabbitMQ broker setup? I mean docker-compose rmq service description if you have one, or a container start bash command.

[stronk7]

Oh, @NelsonNotes, sorry for the lack of answer, I missed your comment completely.

About the code I'm using, I'm not doing anything special. I just have a RabbitBroker instance wrapped into a class named Consumer (also another Producer one) and they both have start() and stop() methods that invoke the RabbitBroker standard (self._broker.start|stop()) and nothing else. They are invoked on app shutdown.

In the good side I must say that I came here because, coincidentally, today I've confirmed that I'm not getting those warnings any more. Don't know if it has been because of some Rabbit update or some FastStream update, for sure my code has not changed.

So with the warnings gone now, maybe this can be closed, unless you consider that there is anything else to check.

Ciao :-)

[Lancetnik]

@stronk7, thank you for coming back! We are happy to know that your problem has been solved. I think we can close this issue for now. Feel free to reopen it if you experience the same problem in the future.