diff --git a/driver/sql/projection_notification_queue.go b/driver/sql/projection_notification_queue.go index a283355..6da8d34 100644 --- a/driver/sql/projection_notification_queue.go +++ b/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 diff --git a/driver/sql/projection_notification_queue_test.go b/driver/sql/projection_notification_queue_test.go new file mode 100644 index 0000000..b80e3e2 --- /dev/null +++ b/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) + } +} diff --git a/go.mod b/go.mod index 91b55ee..5f0751d 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 5e91f8a..11b586d 100644 --- a/go.sum +++ b/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=