Consistency guarantees between publishers to a queue and conditional delete operations on the same queue

[d-shevtsov]

A message gets confirmation (ACK) instead of return (RETURN) due to a race condition between publishing to the direct exchange bound to a single queue and deleting that queue with an empty flag.

The following test scenario describes the issue:

1. A publisher tries to submit 1000 messages to the direct exchange bound to a single queue. If the publisher receives a confirmation, it marks the message as published. If the publisher gets nack/return, it recreates a queue and then retries to submit returned/nacked message until it finally receives a confirmation.
2. A consumer reads the messages from the queue. If the queue gets deleted, it recreates the queue and continues reading messages from the queue. Once a consumer retrieves the message, it marks the message as consumed.
3. A cleanup thread tries to delete the queue only if it is empty (isEmpty flag is set). Publishing happens with a delay so that there are time windows when the queue becomes empty and the thread can delete it.

The experiment waits until all 1000 messages receive confirmed status and then until the queue becomes empty or deleted (so that the consumer has a chance to consume the messages). As the result of the experiment, I expect that all messages are published (and receive confirmations) by the publisher and are received by a consumer. However, few messages receive confirmation but never reach the consumer. The moment they receive ACK is very close to when the queue gets deleted. Therefore I believe there is a race condition between publishing and deleting a queue with the empty flag.

The experiment was done with rabbitmq-java-client version 5.14.2. The same experiment with the spring-amqp library produces the same results. Further, the experiment succeeds (publish and consume all the messages correctly) with RabbitMQ server with version 3.7.20-management and fails for 3.8.0-management, 3.8.23-management, 3.9.16-management, 3.10-rc-management - the issue is likely to be introduced in the 3.8.0 version.

I am attaching the project on which I was able to reproduce the issue for your reference. publishAndDelete.zip

The messages are numbers from 0 to 999. You'll see the similar logs with INFO logger level:

13:24:17.491 INFO  Main Published messages size: 1000, consumed messages size: 964
13:24:17.498 INFO  Main Published but not consumed messages: [8, 74, 174, 191, 198, 202, 217, 229, 302, 309, 324, 332, 339, 343, 384, 404, 407, 423, 424, 452, 453, 474, 479, 482, 497, 523, 544, 559, 637, 722, 737, 829, 832, 957, 985, 998]

With DEBUG level, you'll see particular messages (that were published but did not reach the consumer) receives ACK with a very small time difference from successful queue deletion (search for 'message: ' for all the logs with that message, sometimes message gets return correctly, retries and then gets ack but never reaches a consumer):

13:23:49.681 DEBUG RabbitPublisher Received publish confirm for message: 8
13:23:49.681 DEBUG RabbitQueueCleanup Queue deleted successfully

[olikasg]

This feels like a natural race condition between the async publishing and consuming. Using queue TTL is a better solution for this problem. https://www.rabbitmq.com/ttl.html#queue-ttl
Once the consumer is disconnected from the queue, and no more activity, the queue gets deleted by RabbitMQ.

[d-shevtsov]

This feels like a natural race condition between the async publishing and consuming. Using queue TTL is a better solution for this problem. https://www.rabbitmq.com/ttl.html#queue-ttl Once the consumer is disconnected from the queue, and no more activity, the queue gets deleted by RabbitMQ.

I need to be sure that the message is written to the queue if a publisher gets an acknowledgment with a status ACK. If there is no coordination between sending an acknowledge and queue delete operation, I do not think TTL will behave differently. In essence, TTL is another condition to delete a queue. I believe there will be a chance that two events will happen at the very exact moment:

1. A queue gets deleted according to a TTL mechanism.
2. A publisher that unluckily decided to publish a message at the end of the TTL will get an acknowledgment with a status ACK.

As a result, the message will never reach the queue due to a concurrency issue, but a producer will think everything is correct.

[michaelklishin]

Acknowledgements are sent once the queue moves the message to durable storage. You are talking about a case where the queue is concurrently conditionally deleted.

[michaelklishin]

Sent the previous reply too soon.

It's one thing for a queue to know when a certain delivery is due to be acknowledged. There is no
coordination with any other process involved. It's another
for it to know if a conditional deletion request should immediately proceed, since the conditions can be
"the number of messages in Ready state in the queue" and some of those may be undergoing routing
or delivery in any of the channels in the same virtual host. This would require very expensive
coordination if we were to provide "perfect" (linearizable?) consistency for publishers and conditional deletes.

[d-shevtsov]

Let's focus our discussion on conditional delete of empty queues for now. That is important for me because you are effectively saying that there is no way to safely delete an empty queue (without a message being lost) in RabbitMQ if we expect that a publisher could concurrently publish a message to the same queue.

I can think of two implementations to provide linearizability between a publisher and conditional remove operations:

I. Using a read-write lock.

1. A channel that publishes the message obtains a read lock to write a message to the storage. A read lock can be shared amongst channels, so that publish operations do not block each other.
2. A channel that conditionally deletes an empty queue firstly checks that the queue's storage is empty. Then it tries to acquire an exclusive write lock that can be acquired if publishers do not hold a read lock. If it is successful, it checks the storage once again and proceeds if the storage is empty.

II. Using a message counter and CAS operations on that counter.

1. A channel that publishes the message does CAS operation to increase the counter. If it is successful, it proceeds with the write to the storage. If not, it re-reads the counter and does a CAS once again unless the value is reserved (-1), meaning the queue was removed.
2. A channel that conditionally deletes an empty queue checks that the counter is zero and tries a CAS to the reserved value. If it is successful, then proceed with the delete. If not, then answer that the queue is not empty.

I get you that acquiring a lock for each publish operation is expensive. Maintaining a counter of the messages with CAS operations should be way less expensive and provide consistency at the RabbitMQ level.

[d-shevtsov]

Conditional queue removal worked as expected in RabbitMQ server 3.7.20. Starting from version 3.8.0, it can discard messages by sending ACK and then removing the queue. Is there any chance that the issue was introduced in the 3.8.0 version?

[michaelklishin]

Publishers and consumers do not coordinate, and the publishing party in this scenario (a channel process) does not and should not coordinate with the queue leader. The channel can only guarantee
that basic.publish and related content frames are correctly interleaved and processed in order, together with queue.delete frames. If a different channel (maybe even a different connection)
attempts to concurrently delete the queue, they will not coordinate. Such coordination would be catastrophically expensive, in particular when the channel and the queue leader replica are on different nodes.

And then there's always "in flight" data that has been read from the socket and parsed but hasn't
yet reached the queue. The queue isn't aware of it so it can only use the message counts it has
to determine if conditional deletion should proceed.

This is a practical concurrent/distributed system consistency limitation we are willing to accept.

[d-shevtsov]

First of all, I do not expect any coordination between publishers and consumers. The issue is about coordination between a process that sends an actual acknowledgment to the publisher and a process that handles queue delete operation.

I am not very aware of the internal design of the RabbitMQ, but as far as I get it from your responses (please correct me if I am wrong):

1. A channel process forwards publish / queue delete frames to the queue leader process.
2. A queue leader process is responsible for handling those requests. For publish operation, it initiates acknowledgment and sends it back to the channel. For a queue delete operation, it deletes an underlying queue if conditions are met.
3. A channel and a queue leader process may reside on the different nodes. Coordination between them is therefore expensive.

If the above statements are correct, then a queue leader process is the only decision-maker for both sending acknowledgment for a publish operation and the delete operation, so the coordination between a channel and queue leader is not required. Then, a queue leader can coordinate sending acknowledge and queue delete operations. I expect that the queue leader should respond with ACK to all publish operations before a queue is actually deleted (or marked as deleted). Once the queue is deleted, the queue leader should respond with RETURN.

Further, this behavior was the case in RabbitMQ with 3.7.20. Is there any chance an issue was introduced in the 3.8.0 version?

[michaelklishin]

I don't know what to add. It doesn't matter what the behavior was in 3.7, software evolves over time.
This limitation around consistency/linerizability of these specific events is acceptable to the RabbitMQ core team. It is not worth greatly complicating the implementation over, and definitely not worth the throughput degradation it inevitably will introduce.

[michaelklishin]

If your system is based around conditional removal under a constant message flow through a queue, I'd question that design. Either use long-lived queues, or use the TTL suggestion above, or migrate your publishers so that the queues can be drained and deleted without a constant inflow of new messages.
This would have many parallels to the Blue/Green deployment strategy.

[d-shevtsov]

How will TTL help if there is no coordination between a conditional removal and publish?

My design does not depend on conditional removal. It is merely an optimization to remove unused queues if there was no activity for some period. However, the API provided by the rabbitMQ is misleading as it is clearly said to delete a queue only if it is empty, which assumes coordination/linearizability. If the whole RabbitMQ team does agree that it is not worth implementing such coordination, at least mention that fact in your documentation.

[michaelklishin]

Queue TTL will only deletes queues that have no consumers so it may be a suitable mechanism. It won't help with any "in flight" messages from publishers.

If publishers and consumers must coordinate around something, RabbitMQ is not the right service
for the coordination part. We usually recommend etcd or Consul, which are consistency-oriented and
provide features specifically for service coordination. Some RabbitMQ features can be abused
for coordination but you will quickly hit a wall.