Quorum queues requeue messages to the back of the queue

[mariuszjedrzejewski]

Describe the bug

Tested in Docker Hub images: 3.10.7-management, 3.12.12-management.

NACKed messages are returned on queue, but are tailed to the end of current bunch of undelivered messages. It's not expected behaviour.

Reproduction steps

1. Create queue in any virtual host (including /) of type QUORUM, with argument x-single-active-consumer = true. Let's say queue name is "q_quorum_sac"
2. Connect client (in my example: written in Go - full listing below), with QoS parameters: prefetch_count: 1, prefetch_size: 0, global: false.
   channelRabbitMQ.Qos(1, 0, false)
3. Declare Basic.Consume over created Channel with following parameters (notable ones: exclusive = true, manual ACK in effect):

messages, err := channelRabbitMQ.Consume(
	"q_quorum_sac", // queue name
	"",             // consumer-tag
	false,          // !!! auto-ack !!!
	true,           // exclusive
	true,           // no local
	false,          // no wait
	nil,            // arguments
)

4. Start iterate over incoming messages, NACKing each one received and allow RabbitMQ to requeue the message again:

	go func() {
		for message := range messages {
			// For example, show received message in a console.
			log.Printf(" > Received message: %s\n", message.Body)

			// simulate a processing failure, back message to broker:
			// Negative-ACK for this single message, and requeue
			message.Nack(false, true)

			time.Sleep(1 * time.Second)
		}
	}()

NACKed messages are returned on queue, but are tailed to the end of current undelivered messages. It's not expected behaviour.

Log from application (already 3 published earlier messages, waiting on idle queue before connect to server):

PS C:\Users\XXXXX\GoProjects\test-playground> go run main.go
2024/02/06 09:44:20 Successfully connected to RabbitMQ
2024/02/06 09:44:20 Waiting for messages
2024/02/06 09:44:20  > Received message: quorum_1
2024/02/06 09:44:21  > Received message: quorum_2
2024/02/06 09:44:22  > Received message: quorum_3
2024/02/06 09:44:23  > Received message: quorum_1
2024/02/06 09:44:24  > Received message: quorum_2
2024/02/06 09:44:25  > Received message: quorum_3
2024/02/06 09:44:26  > Received message: quorum_1
2024/02/06 09:44:27  > Received message: quorum_2
2024/02/06 09:44:28  > Received message: quorum_3
exit status 0xc000013a

Expected behavior

NACKed message should return to its place in the queue, i.e. at the beginning. Next message after previously NACKed message should be the same message again, to prevent FIFO order. Delivery of the same message should be continued until message will be positively ACKed by client, as described in https://blog.rabbitmq.com/posts/2020/06/quorum-queues-local-delivery/:

If you need bullet proof FIFO ordering guarantees of ALL messages (including redelivered messages) then you need to use the Single Active Consumer feature with a prefetch of 1. Any redelivered messages get added back to the queue before the next delivery takes place - maintaining FIFO order.

As described above: SAC is used on queue declaration, exclusive consumer is declared, and QoS prefetch count is set to 1.

Additional context

Full code of Go program:

package main

import (
	"log"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	amqpServerURL := "amqp://XXXX:YYYYYYYY@localhost:5672/"

	// Create a new RabbitMQ connection.
	connectRabbitMQ, err := amqp.Dial(amqpServerURL)
	if err != nil {
		panic(err)
	}
	defer connectRabbitMQ.Close()

	// Opening a channel to our RabbitMQ instance over
	// the connection we have already established.
	channelRabbitMQ, err := connectRabbitMQ.Channel()
	if err != nil {
		panic(err)
	}
	defer channelRabbitMQ.Close()

	// prefetch = 1
	channelRabbitMQ.Qos(1, 0, false)

	// Subscribing to queue for getting messages.
	messages, err := channelRabbitMQ.Consume(
		"q_quorum_sac", // queue name
		"",             // consumer-tag
		false,          // !!! auto-ack !!!
		true,           // exclusive
		true,           // no local
		false,          // no wait
		nil,            // arguments
	)
	if err != nil {
		log.Println(err)
	}

	// Build a welcome message.
	log.Println("Successfully connected to RabbitMQ")
	log.Println("Waiting for messages")

	// Make a channel to receive messages into infinite loop.
	forever := make(chan bool)

	go func() {
		for message := range messages {
			// For example, show received message in a console.
			log.Printf(" > Received message: %s\n", message.Body)

			// simulate a processing failure, back message to broker:
			// Negative-ACK for this single message, and requeue
			message.Nack(false, true)

			time.Sleep(1 * time.Second)
		}
	}()

	<-forever
}

Value (true/false) on "Exclusive" flag on Consume() method has no effect if changed.

If queue is declared without "x-single-active-consumer = true" argument - it also has no effect.

Unlike the QUORUM queue, CLASSIC queue works as expected (also 3 messages on queue, the same code used as above, the one and only change is in first parameter of Consume() - queue name). Log:

PS C:\Users\XXXXX\GoProjects\test-playground> go run main.go
2024/02/06 09:39:40 Successfully connected to RabbitMQ
2024/02/06 09:39:40 Waiting for messages
2024/02/06 09:39:40  > Received message: payload1
2024/02/06 09:39:41  > Received message: payload1
2024/02/06 09:39:42  > Received message: payload1
2024/02/06 09:39:43  > Received message: payload1
2024/02/06 09:39:44  > Received message: payload1
exit status 0xc000013a

There was 3 messages with "payload1", "payload2" and "payload3" payloads.
Delivery of "payload1" was NACKed and redelivered, and next delivery is again "payload1", until ACKed - it's expected behaviour.

Tcpdump is captured, showing that client is sending NACKs, but server performs delivery of next message in response:

Client is setting QoS:

Client is declaring of consuming messages from q_quorum_sac queue:

RabbitMQ delivers first message with "quorumsac1" payload (delivery tag: 1):

Client NACK for this delivery tag:

RabbitMQ sends next message with "quorumsac2":

[michaelklishin]

NACKed messages are returned on queue, but are tailed to the end of current undelivered messages.
It's not expected behaviour.

This is a documented behavior and not a bug. Quorum queues cannot keep accumulating messages at the head of the queue, because in this case they cannot move on to truncate the log, eventually running the node out of disk space.

This fairly common scenario and a fundamental problem where QQs cannot free up disk space because of a certain sequence of (intentional or not) client operations is why consumer delivery timeout was introduced. These two changes make quorum queues a lot more resilient in the face of certain (usually repeated) operations performed by applications with curious design decisions (around how they handle deliveries and acknowledge them).

[mariuszjedrzejewski]

Ok, just read about repeated requeues, will try to check it. Thanks for pointing out this one.

[mariuszjedrzejewski]

@michaelklishin, thank you!

PS> go run main.go
2024/02/06 14:17:51 Successfully connected to RabbitMQ
2024/02/06 14:17:51 Waiting for messages
2024/02/06 14:17:51  > Received message: PL1
2024/02/06 14:17:52  > Received message: PL1
2024/02/06 14:17:53  > Received message: PL1
2024/02/06 14:17:54  > Received message: PL1
2024/02/06 14:17:55  > Received message: PL1
2024/02/06 14:17:56  > Received message: PL1
2024/02/06 14:17:57  > Received message: PL1
2024/02/06 14:17:58  > Received message: PL1
2024/02/06 14:17:59  > Received message: PL1
2024/02/06 14:18:00  > Received message: PL1
2024/02/06 14:18:01  > Received message: PL1
2024/02/06 14:18:02  > Received message: PL1
2024/02/06 14:18:03  > Received message: PL1
exit status 0xc000013a

Now message is back on top and redelivered until NACked or exceed the value of x-delivery-limit. That's what I want. Thanks again!

[michaelklishin]

Well, that's an interesting way to solve it (I honestly could not think of it) but as long as you understand this behavior, why it exists and how to work around it, it's all good 👏

[michaelklishin]

In general, the idea of "FIFO" when you requeue messages and have multiple (concurrent, competing) consumers becomes a moot point. Deliveries will effectively "jump" their queue position in such cases, even if you requeue to the head, because of the competing consumer scenario, QoS allowing for multiple deliveries at a time, and so on. In the presence of concurrent modifications, FIFO becomes FIFO-ish.

Therefore the classic queue behavior of (trying it) requeue to the original position in the head is mostly an illusion with a lot of workloads.