From 385347b53b38f7da4e012b03c11d02f0602b7904 Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Fri, 18 Sep 2026 10:33:18 +0000 Subject: [PATCH 1/2] Bump github.com/rabbitmq/amqp091-go from 1.13.0 to 1.15.0 Bumps [github.com/rabbitmq/amqp091-go](https://github.com/rabbitmq/amqp091-go) from 1.13.0 to 1.15.0. - [Release notes](https://github.com/rabbitmq/amqp091-go/releases) - [Changelog](https://github.com/rabbitmq/amqp091-go/blob/main/CHANGELOG.md) - [Commits](https://github.com/rabbitmq/amqp091-go/compare/v1.13.0...v1.15.0) --- updated-dependencies: - dependency-name: github.com/rabbitmq/amqp091-go dependency-version: 1.15.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 9777772..b90a80e 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.9.4 github.com/stretchr/testify v1.12.1 go.uber.org/zap v1.28.0 diff --git a/go.sum b/go.sum index 0379809..d9d053c 100644 --- a/go.sum +++ b/go.sum @@ -193,8 +193,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= From 66205ee8c361eaccae8be6e6e825381bd465bb47 Mon Sep 17 00:00:00 2001 From: Richard Palm Date: Mon, 12 Oct 2026 14:08:19 +0200 Subject: [PATCH 2/2] Stop processors instead of handling phantom nil notifications on shutdown When the notification processor is stopped, the done channel is closed followed by the queue channel. A processor waiting in Next could then randomly receive the zero value of the closed queue channel and mistake it for a real nil notification, running the ProcessHandler during shutdown. Every out-of-sync notification the handler tried to queue then failed with "processor was stopped", which surfaced as intermittent "failed to queue notification" errors in the aggregate projector integration tests. Next now checks whether the queue was closed after each receive and always reports the processor as stopped, so no handler runs after shutdown begins. A regression test covers both the shutdown and the normal nil-delivery paths. Assisted-by: Crush:zai-org/GLM-5.3 --- driver/sql/projection_notification_queue.go | 11 ++++ .../sql/projection_notification_queue_test.go | 51 +++++++++++++++++++ 2 files changed, 62 insertions(+) create mode 100644 driver/sql/projection_notification_queue_test.go 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) + } +}