Bug about confirm mode and r eturn mode

[wushu037]

Describe the bug

Thanks for reading, it could be my problem, but I have been troubleshooting for a long time without finding the cause.
I am using go1.18 and github.com/rabbitmq/amqp091-go v1.8.1

Here is my source code, I'm doing a practice on confirm mode and return mode
I refer to https://www.rabbitmq.com/confirms.html#publisher-confirms
I ran into several problems and commented them out in the code, here are the statistics

1. confirm mode: can't use for range to receive callbacks for posting failures
2. confirm mode: using <-, I can receive the same failed callback countless times
3. return mode: please forgive me for not finding a description of how to turn it on in the documentation. I'm not sure if it will have the same problem as confirm mode even if it is enabled

Reproduction steps

1. About "confirm mode"

package main

import (
	"context"
	"github.com/rabbitmq/amqp091-go"
	"log"
	"time"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Panicf("%s: %s", msg, err)
	}
}

func main() {
	conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	// open confirm mode
	err = ch.Confirm(false)
	failOnError(err, "Failed to open confirm mode")

	// NotifyPublish
	confirmNotice := ch.NotifyPublish(make(chan amqp091.Confirmation, 1))
	go func() {
		notice := <-confirmNotice
		log.Printf("【<-】confirm notice: %v", notice)
		// ⭐️ for range does not fetch the data in chan, but it clearly has data
		for notice := range confirmNotice {
			log.Printf("【range】confirm notice: %v", notice)
		}
		// ⭐️ It can fetch the same acknowledgement countless times
		for i := 0; i < 20; i++ {
			notice = <-confirmNotice
			log.Printf("【<-】confirm notice: %v", notice)
		}
	}()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	body := "Hello World!"
	// Publish a message to a non-existent switch, it should return false to `NotifyPublish` once
	err = ch.PublishWithContext(ctx,
		"xxxyyy", // exchange
		"",       // routing key
		false,    // mandatory
		false,    // immediate
		amqp091.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		})
	failOnError(err, "Failed to publish a message")
	log.Printf(" [x] Sent %s", body)

	var forever chan struct{}
	<-forever
}

2. About "return mode"

	// I can't find where to open return mode

	// NotifyReturn
	returnNotice := ch.NotifyReturn(make(chan amqp091.Return, 1))
	go func() {
		for notice := range returnNotice {
			log.Printf("return notice: %v", notice)
		}
	}()

	// To publish a message to a non-existent queue, it should return false once to `NotifyReturn`.
	err = ch.PublishWithContext(ctx,
		"",    // exchange
		"zzz", // routing key
		false, // mandatory
		false, // immediate
		amqp091.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		})
	failOnError(err, "Failed to publish a message")
	log.Printf(" [x] Sent %s", body)

Expected behavior

I hope to get a solution to this problem, thanks!

[wushu037]

hi, @lukebakken
Has this bug been verified? I would like to know the validity of this issue.

[Zerpet]

Hi @wushu037,

I can confirm that there is no bug, the code is working as intended. I don't know what are the requirements for your practice; I guess you have to demonstrate publish confirm and return (see below).

For when you will receive a publish confirmation, I'm going to quote the documentation:

For unroutable messages, the broker will issue a confirm once the exchange verifies a message won't route to any queue [...]

For routable messages, the basic.ack is sent when a message has been accepted by all the queues. For persistent messages routed to durable queues, this means persisting to disk. [...]

In your source code, the PublishWithContext function suceeds (i.e. sends the message successfully), but RabbitMQ realises that the exchange xxxyyy does not exist. This is a channel exception, according to the AMQP protocol. Since there is a channel exception, the channel is closed; when the library detects the channel exception it closes the (AMQP) channel, and the confirmNotice (Golang) channel.

You are able to receive from the channel because Golang's receive operator always returns on a closed channel. The for range loop works, and it stopped as soon as the channel was closed (as intended).

In order to get a basic.return, you have to flag your publish with mandatory or immediate (or both).

Here is your source code modified to demonstrate publish confirm, return and channel exception.

issue196.go

package main

import (
	"context"
	"github.com/rabbitmq/amqp091-go"
	"log"
	"time"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Panicf("%s: %s", msg, err)
	}
}

func main() {
	conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	// open confirm mode
	err = ch.Confirm(false)
	failOnError(err, "Failed to open confirm mode")

	// NotifyPublish
	confirmNotice := ch.NotifyPublish(make(chan amqp091.Confirmation, 1))
	go func() {
		// ⭐️ for range does fetch the data in chan :-)
		for notice := range confirmNotice {
			// publish confirms that the server received the message
			// it does NOT mean that the message was routed to a queue
			log.Printf("【range】confirm notice: %v", notice)
		}
		// ⭐️ channel is closed due to a channel exception (publish to non-existing exchange)
		// you can receive from the channel, but you will get the zero-value of Confirmation
		for i := 0; i < 20; i++ {
			notice, ok := <-confirmNotice
			log.Printf("【<-】confirm notice (ok: %v): %v", ok, notice)
		}
	}()

	// NotifyReturn
	returnNotice := ch.NotifyReturn(make(chan amqp091.Return, 1))
	go func() {
		for notice := range returnNotice {
			log.Printf("【range】return notice: %v", notice)
		}
	}()

	// NotifyClose
	channelNotice := ch.NotifyClose(make(chan *amqp091.Error, 1))
	go func() {
		select {
		case notice := <-channelNotice:
			log.Printf("【select】channel notice: %v", notice)
		}
	}()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	body := "Hello World!"
	// exchange exists, but message is undeliverable because there is no binding to route the message
	// it will notify return because it has mandatory flag
	err = ch.PublishWithContext(ctx,
		"",                    // exchange
		"this-does-not-exist", // routing key
		true,                  // mandatory
		false,                 // immediate
		amqp091.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		})
	failOnError(err, "Failed to publish a message")
	log.Printf(" [x] Sent %s", body)

	// Publish a message to a non-existent switch, it will cause a channel exception,
	// close the channel and notify channel close
	err = ch.PublishWithContext(ctx,
		"xxxyyy", // exchange
		"",       // routing key
		false,    // mandatory
		false,    // immediate
		amqp091.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		})
	failOnError(err, "Failed to publish a message")
	log.Printf(" [x] Sent %s", body)

	var forever chan struct{}
	<-forever
}

Edit: RabbitMQ does not support immediate flag.

[wushu037]

Hi @Zerpet ,
So that's it, it was the channel exception that caused the problem. Thanks for the guidance, I finished my practice.

Also I found that rabbitmq 3.0 deprecates immediate and when setting it to true, NotifyClose() gets Exception (540) Reason: "NOT_IMPLEMENTED - immediate=true". So should amqp091-go adjust the PublishWithContext comment for immediate?

[Zerpet]

This library implements the AMQP protocol, which has support for the immediate flag. I don't want to make modifications specific to RabbitMQ, since this client could be used by other AMQP message brokers, like LavinMQ, and those brokers may support the immediate flag. In addition to that, I believe the error message makes it very clear what the problem is, and how to correct it.

[wushu037]

Okay, thanks!