Bump amqp091-go to 1.15.0 and fix projector shutdown race

driver/sql/projection_notification_queue.go

@@ -80,6 +80,17 @@ func (nq *NotificationQueue) Next(ctx context.Context) (*ProjectionNotification,
 		case <-ctx.Done():
 			return nil, true
 		case notification := <-nq.queue:
+			// Receiving from a closed queue channel returns the zero value (nil). This must
+			// not be handled as a notification, as doing so would trigger the ProcessHandler
+			// for a nil notification while the processor is shutting down, causing the
+			// handler to fail queueing out-of-sync notifications. Once the queue was closed
+			// the processor should always be stopped.
+			select {
+			case <-nq.done:
+				return nil, true
+			default:
+			}
+
 			if notification != nil && notification.ValidAfter.After(time.Now()) {
 				nq.queueNotification(notification)
 				continue

driver/sql/projection_notification_queue_test.go

@@ -0,0 +1,51 @@
+//go:build unit
+
+package sql
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNotificationQueueDeliversNotifications(t *testing.T) {
+	nq := newNotificationQueue(10, 0, NopMetrics)
+	queueClose := nq.Open()
+	defer queueClose()
+
+	ctx := context.Background()
+
+	// A queued nil notification is a valid trigger for out-of-sync projections
+	require.NoError(t, nq.Queue(ctx, nil))
+	notification, stopped := nq.Next(ctx)
+	assert.False(t, stopped)
+	assert.Nil(t, notification)
+
+	// A queued notification is delivered to the caller
+	queued := &ProjectionNotification{No: 1, AggregateID: "abc"}
+	require.NoError(t, nq.Queue(ctx, queued))
+	notification, stopped = nq.Next(ctx)
+	assert.False(t, stopped)
+	assert.Equal(t, queued, notification)
+}
+
+func TestNotificationQueueStopsAfterClose(t *testing.T) {
+	nq := newNotificationQueue(10, 0, NopMetrics)
+	queueClose := nq.Open()
+
+	ctx := context.Background()
+
+	// Mimic the shutdown performed by the processor: the done channel is closed
+	// followed by the queue channel. A processor observing this state must stop
+	// and never receive the zero value of the closed queue channel as a phantom
+	// nil notification.
+	queueClose()
+
+	for i := 0; i < 100; i++ {
+		notification, stopped := nq.Next(ctx)
+		assert.True(t, stopped, "Next must signal that the processor should stop")
+		assert.Nil(t, notification)
+	}
+}

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/lib/pq v1.10.9
 	github.com/mailru/easyjson v0.9.2
 	github.com/prometheus/client_golang v1.13.0
-	github.com/rabbitmq/amqp091-go v1.13.0
+	github.com/rabbitmq/amqp091-go v1.15.0
 	github.com/sirupsen/logrus v1.10.2
 	github.com/stretchr/testify v1.12.1
 	go.uber.org/zap v1.28.0

go.sum

@@ -194,8 +194,8 @@ github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1
 github.com/prometheus/procfs v0.7.3/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
 github.com/prometheus/procfs v0.8.0 h1:ODq8ZFEaYeCaZOJlZZdJA2AbQR98dSHSM1KW/You5mo=
 github.com/prometheus/procfs v0.8.0/go.mod h1:z7EfXMXOkbkqb9IINtpCn86r/to3BnA0uaxHdg830/4=
-github.com/rabbitmq/amqp091-go v1.13.0 h1:L8NA1WtF76C6KA3LAoufjfLgbist/If1UQYcsOjtxXA=
-github.com/rabbitmq/amqp091-go v1.13.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
+github.com/rabbitmq/amqp091-go v1.15.0 h1:LEQL4/yp48/Wigt6A6XOu18RQRo8ZHtB5I/KZJn+gkw=
+github.com/rabbitmq/amqp091-go v1.15.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=