BlockingConnection.process_data_events does not account for channel closure

[lfse-slafleur]

Hi all,

Thanks for building and maintaining pika!

We followed this example in building an app using pika: https://github.com/pika/pika/blob/main/examples/long_running_publisher.py
The function on line 25 self.connection.process_data_events(time_limit=1) blocks a thread to handle all AMQP processing.

We are communicating with RabbitMQ and we ran into an edge case scenario where a job took so long to acknowledge a message it triggered the consumer_timeout which caused the channel to be closed, but the connection remained open. The pika app proceeded to silently die as the channel closure did not propagate as a pika.exceptions.ChannelClosedByBroker.

Upon code inspection BlockingChannel._process_data_events does add this functionality where it calls BlockingConnection.process_data_events and also checks if the channel had closed and raises raise BlockingChannel._closing_reason if it exists.

Steps to reproduce:

• Spin up RabbitMQ with consumer_timeout = 60
• Run https://github.com/pika/pika/blob/main/examples/long_running_publisher.py with auto_ack off and, on a separate thread, sleep for 120 seconds when a message is received before acknowledging the message to RabbitMQ (using the thread-safe callback).
• Note: RabbitMQ now closed the channel with Channel.Close reason: PRECONDITION_FAILED - delivery acknowledgement on channel 1 timed out. Timeout value used: 60000 ms. This timeout value can be configured, see consumers doc guide to learn more
• Note: The pika app keeps blocking on self.connection.process_data_events(time_limit=1) as the connection is still opened but there is no activity as the channel is closed. The app is not notified the channel has closed.

What I expected to happen:
pika.exceptions.ChannelClosedByBroker should be thrown when blocking on BlockingConnection.process_data_events. From the documentation I read this method is the one a pika app should block on.

Options how to resolve (from my limited understanding on the code base):

1. Make BlockingChannel._process_data_events public instead of private.
2. Allow BlockingChannel._closing_reason to be inspected through a BlockingChannel.get_closing_reason so the app is able to check itself if the channel has closed and why.
3. Propagate BlockingChannel._closing_reason for any channel when blocking on BlockingConnection.process_data_events. Perhaps this option makes less sense as the BlockingChannel currently is layered on top of BlockingConnection.
4. Change the example in https://github.com/pika/pika/blob/main/examples/long_running_publisher.py to show how to properly block so channel closure is also propagated to the example.

Unclear challenges:

• With option 1 there is no clear path how to block with a single thread on all active channels.

If there are any comments or questions please let me know. I tried to do my due diligence in making sure this is a bug and not just a question/discussion.

[lukebakken]

If your application takes more than a few milliseconds to process a message, you should process it on a separate thread.

Sleeping for two minutes blocks Pika's I/O completely and is a misuse of the library.

Pika is not thread-safe.

[lfse-slafleur]

@lukebakken Thanks for the quick response. I have added the necessity to use a separate thread in the description for completeness but that was not the point of my ticket. The point is that the acknowledgement of a message to RabbitMQ is postponed until after the consumer_timeout at which point RabbitMQ closes the channel but not the connection to the pika app. The problem is that the pika.exceptions.ChannelClosedByBroker is not thrown and no other method seems to be available to signal the app that the channel has closed.

[lukebakken]

The problem is that the pika.exceptions.ChannelClosedByBroker is not thrown and no other method seems to be available to signal the app that the channel has closed.

If you correctly use a separate thread, does this problem still exist? Please provide code to reproduce, it will help me out. I am very busy working on the next .NET client release and don't have much extra time to look at this at this point.

[lfse-slafleur]

@lukebakken Yep not a problem. Here is a codified example of the issue: https://github.com/TNO-SlaFleur/pika-channel-closed-test

You can run it by running run.sh and it will deploy rabbitmq and the test app using docker compose. The log/print lines should tell the story.

On line 61 you can see the self.connection.process_data_events(time_limit=1) call that I would expect to throw a ChannelClosedByBroker.

The test app uses a separate read & write channel to simulate a separate read app and write app. Did not have the time to fully split it into 2 dockers/processes/threads. But hopefully the point is clear. I would expect self.connection.process_data_events(time_limit=1) to be the (only) blocking call for the app to block on to handle all channels and everything on top of that connection.

[lukebakken]

Thank you very much for the code. I will investigate when I have time.

[lfse-slafleur]

Not a problem. Least I can do to contribute to this awesome opensource project. Good luck on the .NET client release!

For anyone reading this who are in a similar situation there is a workaround:

Assuming you only have a single channel on your connection (which is quite common) you can use <reference to blocking channel object>._process_data_events(time_limit=1) instead of <reference to blocking connection>.process_data_events(time_limit=1). This does not mean there aren't other workarounds, but this is working successfully for me at the moment. The reason it works is because BlockingChannel._process_data_events calls BlockingConnection.process_data_events and adds the ChannelClosedByBroker exception. It does not add anything else for pika 1.3.2. As it is a private function (starts with _), this is not an official/correct solution.

[lukebakken]

You know it seems like you really understand this issue - feel free to open a PR with a fix that you think is appropriate. I appreciate it.