Call to Delivery.Ack blocks indefinitely in case of disconnection

[corcus]

Steps to reproduce

1. Create a queue in the RabbitMQ server
2. Call consume and start consuming messages
3. While processing a message and BEFORE Delivery.Ack is called disconnect from the RabbitMQ server (for example turn off your internet)

The call to Delivery.Ack blocks indefinitely instead of returning an error.

[DanielePalaia]

Hi, this issue is a bit old, are you still experimenting this problem? I'm trying to reproduce the issue but till now I wasn't really able to. I tried to create a sort of unit test, Consuming message and closing connection (I tried from the rabbitMQ UI, turning of internet and stopping the broker) while consuming and before Delivery.Ack is called, but Ack is returning an error as expected.
I will try to investigate in the code in the meantime.

This is the code I was testing:

func TestShouldNotWaitOnDeliveryAckAfterConnectionCloseIssue19(t *testing.T) {
	connPublish := integrationConnection(t, "TestShouldNotWaitOnDeliveryAckAfterConnectionCloseIssue19")
	chPublish, err := connPublish.Channel()
	if err != nil {
		t.Fatalf("channel error: %v", err)
	}

	exchangeName := "exchangeTestIssue19"
	queueName := "queueTestIssue19"
	nMsg := 10000

	if err := chPublish.ExchangeDeclare(exchangeName, "fanout", false, true, false, false, nil); err != nil {
		t.Fatalf("cannot declare %v: got: %v", exchangeName, err)
	}

	if _, err := chPublish.QueueDeclare(queueName, false, true, false, false, nil); err != nil {
		t.Fatalf("cannot declare %v: got: %v", queueName, err)
	}

	if err := chPublish.QueueBind(queueName, "#", exchangeName, false, nil); err != nil {
		t.Fatalf("cannot bind %v to %v: got: %v", queueName, exchangeName, err)
	}

	//Publish one element in
	for i := 0; i < nMsg; i++ {
		err = chPublish.Publish(exchangeName, "#", false, false, Publishing{Body: []byte("abc")})
		if err != nil {
			t.Fatalf("PublishWithDeferredConfirm error: %v", err)
		}
	}

	chPublish.Close()
	connPublish.Close()

	connConsume := integrationConnection(t, "TestShouldNotWaitOnDeliveryAckAfterConnectionCloseIssue19")
	chConsume, err := connConsume.Channel()
	if err != nil {
		t.Fatalf("channel error: %v", err)
	}

	msgs, err := chConsume.Consume(queueName, "#", false, false, false, false, nil)
	if err != nil {
		t.Fatalf("Could not consume")
	}

	i := 0

	for d := range msgs {
		i++
		t.Logf(
			"got %dB delivery: [%v] %q",
			len(d.Body),
			d.DeliveryTag,
			d.Body,
		)

		// Simulate a network issue (close connection from management-api, ecc...)
		err := d.Ack(true)

		if err == nil {
			t.Logf("Ack should return a network issue %v", err)
		}

	}

	t.Logf("ending %v", err)

}

[corcus]

Hey thanks for looking into this. It has been a long time and I don't remember exactly how this issue came up.
To the best of my memory it happened like this.

In the range loop consuming the messages there was some long running task (or a breakpoint during debugging)

for d := range msgs {
		i++
		t.Logf(
			"got %dB delivery: [%v] %q",
			len(d.Body),
			d.DeliveryTag,
			d.Body,
		)

		// Simulate a network issue (close connection from management-api, ecc...)
		err := d.Ack(true)

		if err == nil {
			t.Logf("Ack should return a network issue %v", err)
		}

	}

Then while the long running task is running or while the breakpoint is hit, disconnect from the internet and allow the long running task to complete (or move on from the breakpoint) to the Ack part.
That's how I remember the issue happening.

[DanielePalaia]

HI @corcus I tried to put a sleep in place of breakpoint or to simulate an heavy load with a loop, then close the connection from UI, and continue with the Ack, but I wasn't really able to reproduce it. Let's close this one as now old, in case it will happen again to you please open a new one maybe with a trace if you are able to get it. Thanks!

[DanielePalaia]

As at the moment not reproducible.