[watermill-amqp] Data race on ConnectionWrapper.connected between IsConnected() and handleConnectionClose()

[watermill-amqp] Data race on ConnectionWrapper.connected between IsConnected() and handleConnectionClose()

Filing here since issues are turned off on the watermill-amqp repo.

Hit this on watermill-amqp v3.1.0 with a readiness probe that calls IsConnected(). Any broker blip makes -race fire, so our integration tests can't run with the detector on.

handleConnectionClose reassigns c.connected on both arms:

	case <-c.closing:
		c.connected = make(chan struct{})   // connection.go:136
		return
	case err := <-notifyCloseConnection:
		c.connected = make(chan struct{})   // connection.go:139
		c.reconnect()

and IsConnected() selects on it (connection.go:114) with nothing in between.

What made me think this is an oversight rather than intentional: connect() already takes amqpConnectionLock when it writes that same field at :96, so the two writers disagree about whether the field is guarded. Connection() at :105 reads amqpConnection unlocked too, and that one is written under the lock at :96 as well.

Repro

docker run --rm -p 5672:5672 rabbitmq:4

package main

import (
	"log"
	"time"

	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill-amqp/v3/pkg/amqp"
)

func main() {
	cfg := amqp.NewDurableQueueConfig("amqp://guest:guest@localhost:5672/")
	conn, err := amqp.NewConnection(cfg.Connection, watermill.NopLogger{})
	if err != nil {
		log.Fatal(err)
	}

	done := make(chan struct{})
	go func() {
		defer close(done)
		deadline := time.Now().Add(3 * time.Second)
		for time.Now().Before(deadline) {
			_ = conn.IsConnected()
		}
	}()

	// drop the TCP connection so NotifyClose fires and we take the reconnect arm
	if err := conn.Connection().Close(); err != nil {
		log.Fatal(err)
	}
	<-done
}

go run -race .:

WARNING: DATA RACE
Write at 0x00c0000c6048 by goroutine 13:
  ...amqp.(*ConnectionWrapper).handleConnectionClose()
      pkg/amqp/connection.go:139
  ...amqp.NewConnection.gowrap1()
      pkg/amqp/connection.go:48

Previous read at 0x00c0000c6048 by goroutine 14:
  ...amqp.(*ConnectionWrapper).IsConnected()
      pkg/amqp/connection.go:114
  main.main.func1()
      main.go:23

Reproduces on darwin/arm64 and linux/amd64, go1.26.

Calling Close() instead of dropping the connection gives the same race via :136, but the reconnect arm seemed worth reporting since it happens without the application doing anything.

Taking amqpConnectionLock around the two writes and around the reads in IsConnected()/Connection() would do it, unless you'd rather not put a lock in accessors that get polled, in which case an atomic.Pointer for connected works too. Glad to put up a PR if you tell me which you'd prefer.