No exception thrown on channel exception from basic.{publish,ack,nack}

[awons-daft]

Consider the following example:

$connection = new AMQPStreamConnection('localhost', 5672, 'user', 'password');
$channel = $connection->channel();

$channel->basic_publish(new AMQPMessage('message 1'), 'undefined_exchange', 'routing_key'); // here the exception is reported to the channel and channel is closed on RabbitMQ side
$channel->basic_publish(new AMQPMessage('message 2'), 'defined_exchange', 'routing_key'); // this message, even though valid, will never reach its destination

The defined_exchange already exists and the undefined_exchange does not.
First call to basic_publish will result in channel exception being reported to the channel by RabbitMQ and channel being closed. The issue I see here is that in userland we have no clue that this has happened and keep on sending new messages. But they will never arrive to the server because channel has already been closed.

The same will happen when you have a consumer defined with prefetch grater than 1. Take a look at the following example:

$connection = new AMQPStreamConnection('localhost', 5672, 'user', 'password');
$channel = $connection->channel();
$channel->basic_qos(null, 2, null);

$message = $channel->basic_get('my_queue');
$channel->basic_ack('invalid_tag'); // here the exception is reported to the channel and channel is closed on RabbitMQ side

$message = $channel->basic_get('my_queue'); // we still get this message because it is already in buffer
$channel->basic_ack($message->delivery_info['delivery_tag']); // This message will not actually be acknowledged

First message is processed and ack-ed with an invalid delivery_tag. This results in exception being reported to the channel and channel being closed (on the server side). Due to QOS settings there are already other messages in the stream buffer so the library will simply grab the next available message without realizing there was an exception on the channel. The library will eventually figure it out but only after all messages have been already read from the buffer and the next frame that is there, is the exception itself.

Both of the behaviors can be dangerous.
First one because your well defined messages will never reach their destination without any apparent reason.
Second one (even though very unlikely) can lead to processing some messages twice when we definitely need to process them only once.

This behavior also presents itself when people are complaining that there is an exception being raised when they already try to close the connection. This is exactly the same problem. We send close message to the server but instead of getting the response we get previously buffered exception from already closed channel.

As a solution would be to throw an exception as soon as it is possible to detect there is an exception waiting in a buffer. I do understand that this may impact performance to some extend but it would provide a "logical" behavior.

Tested on php-amqplib/php-amqplib v2.6.3.

[michaelklishin]

basic.ack and basic.nack do not have response in the protocol, therefore clients that wait for it before throwing an exception cannot do the same thing here. This is a significant difference from, say, queue.declare. Some clients choose to provide a callback (not very elegant), others rely on the fact that subsequent operations will fail because the channel is closed.

This protocol design decision is likely intentional: all methods that have no response are on the hot path (basic.publish, basic.ack, basic.reject, and basic.nack).

As a solution would be to throw an exception as soon as it is possible to detect there is an exception waiting in a buffer

This sounds worth investigating.

[awons-daft]

I do see your point here. I understand that AMQP protocol works this way.
But the protocol also states that in case of an error an exception is reported on a channel and it gets closed.
So I still think this is a bug in a library itself that deliberately decides to ignore that fact and still allows to publish new messages. I see this as two problems:

1. On publisher end - if you try to push to a channel on which an exception has been reported it should immediately lead to exception being thrown.

2. On consumer end - if you try to consume a message (especially with prefetch grater than 1) form a channel on which an exception has been reported it should result in exception being thrown (even when the exception is at the end or in the middle of a buffer). Otherwise you allow for operating on an already closed channel as if nothing happened.

To sum it up - I think the exception from a channel should be raised to userland code as soon as it appears in the buffer. Regardless on what operation we are trying to perform. We know the channel is closed so why would we allow to operate on it anyway? What would we potentially gain here?

Does it make any sense what I say?

[Patrick-Remy]

How can this be handled? I have some long-running consumer processes, where based on queue-input, many different publishes are done. If a single channel error (e.g. due to unknown exchange like in this example) occurs, the consumer will silently lose all subsequent publishes.