From f6858fa6130b8f252b3c8010a31fab0ad99cc443 Mon Sep 17 00:00:00 2001 From: sergeyb Date: Mon, 9 Nov 2026 21:53:10 +0000 Subject: [PATCH 1/2] fix(consumer): retain invalid final-queue envelopes Summary: Keep invalid envelope identities durable for subscriptions with unlimited retries and no further dead-letter queue. The pre-controller guard uses the registry's actual subscription policy to Nack these deliveries instead of Reject, while finite primary subscriptions keep their existing behavior. Test Plan: TestConsumer_ProcessDelivery_InvalidTenantMetadata covers final retention, finite primary rejection, and settlement failures. The consumer, SQL queue, and existing consumer integration Bazel targets verify the extracted change independently. Revert Plan: Revert the dependent gateway recovery change before this policy; leave retained final-queue rows intact. --- platform/consumer/consumer.go | 48 +++++++++++------ platform/consumer/consumer_test.go | 86 ++++++++++++++++-------------- 2 files changed, 77 insertions(+), 57 deletions(-) diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go index 7aab1b906..4dcb4b4b6 100644 --- a/platform/consumer/consumer.go +++ b/platform/consumer/consumer.go @@ -308,7 +308,7 @@ func (m *consumer) consumeLoop(ctx context.Context, controller Controller, topic m.shutdownPartitions(partitionChs, &wg) return } - if m.rejectDeliveryWithInvalidTenant(ctx, controller, delivery) { + if m.rejectDeliveryWithInvalidTenant(ctx, controller, delivery, controllerScope) { continue } @@ -340,27 +340,41 @@ func (m *consumer) consumeLoop(ctx context.Context, controller Controller, topic } } -func (m *consumer) rejectDeliveryWithInvalidTenant(ctx context.Context, controller Controller, delivery extqueue.Delivery) bool { +func (m *consumer) rejectDeliveryWithInvalidTenant(ctx context.Context, controller Controller, delivery extqueue.Delivery, scope tally.Scope) bool { msg := delivery.Message() - if err := entityqueue.ValidateTenantMetadata(msg); err != nil { - m.logger.Errorw("rejecting message with inconsistent queue identity", - "controller", controller.Name(), + err := entityqueue.ValidateTenantMetadata(msg) + if err == nil { + return false + } + + settle := delivery.Reject + action := "reject" + config, ok := m.registry.SubscriptionConfig(controller.TopicKey(), controller.ConsumerGroup()) + if ok && config.Retry.MaxAttempts == 0 && !config.DLQ.Enabled { + // Reject without a DLQ acknowledges the message, violating final recovery's retention policy. + settle = delivery.Nack + action = "nack" + } + metrics.NamedCounter(scope, "envelope", "invalid_identity", 1) + m.logger.Errorw("message has inconsistent queue identity", + "controller", controller.Name(), + "topic_key", controller.TopicKey(), + "consumer_group", controller.ConsumerGroup(), + "message_id", msg.ID, + "tenant", msg.Tenant, + "action", action, + "error", err, + ) + if settleErr := settle(ctx, failure.New(err.Error())); settleErr != nil { + m.logger.Errorw("failed to settle message with inconsistent queue identity", "topic_key", controller.TopicKey(), + "consumer_group", controller.ConsumerGroup(), "message_id", msg.ID, - "tenant", msg.Tenant, - "error", err, + "action", action, + "error", settleErr, ) - if rejectErr := delivery.Reject(ctx, failure.New(err.Error())); rejectErr != nil { - m.logger.Errorw("failed to reject message with inconsistent queue identity", - "controller", controller.Name(), - "topic_key", controller.TopicKey(), - "message_id", msg.ID, - "error", rejectErr, - ) - } - return true } - return false + return true } // shutdownPartitions closes all partition channels to signal processPartition diff --git a/platform/consumer/consumer_test.go b/platform/consumer/consumer_test.go index 61de23d91..b95d70373 100644 --- a/platform/consumer/consumer_test.go +++ b/platform/consumer/consumer_test.go @@ -332,46 +332,52 @@ func TestConsumer_ProcessDelivery_Success(t *testing.T) { require.NoError(t, err) } -func TestConsumer_ProcessDelivery_RejectsConflictingTenantMetadata(t *testing.T) { - ctrl := gomock.NewController(t) - deliveryChan := make(chan extqueue.Delivery, 1) - mockSub := queuemock.NewMockSubscriber(ctrl) - mockSub.EXPECT().Subscribe(gomock.Any(), gomock.Any(), gomock.Any()).Return(deliveryChan, nil) - mockQ := queuemock.NewMockQueue(ctrl) - mockQ.EXPECT().Subscriber().Return(mockSub) - - c := New( - zaptest.NewLogger(t).Sugar(), - tally.NoopScope, - newRegistry(t, mockQ, testTopicKeyStart, "test-group"), - errs.NewClassifierProcessor(), - consumergatenoop.New(), - ) - handler := &testController{} - setupController(handler, "test-handler", testTopicKeyStart, "test-group", - func(context.Context, Delivery) error { - assert.Fail(t, "controller must not receive inconsistent delivery identity") - return nil - }, - ) - require.NoError(t, c.Register(handler)) - require.NoError(t, c.Start(context.Background())) - - msg := entityqueue.NewMessage("mismatch", []byte("payload"), "partition1", map[string]string{ - entityqueue.MetadataKeyQueueName: "tenant-b", - }) - msg.Tenant = "tenant-a" - rejected := make(chan struct{}) - delivery := queuemock.NewMockDelivery(ctrl) - delivery.EXPECT().Message().Return(msg).AnyTimes() - delivery.EXPECT().Reject(gomock.Any(), gomock.Any()).DoAndReturn(func(context.Context, failure.Failure) error { - close(rejected) - return nil - }) - - deliveryChan <- delivery - <-rejected - require.NoError(t, c.Stop(30000)) +func TestConsumer_ProcessDelivery_InvalidTenantMetadata(t *testing.T) { + for _, final := range []bool{false, true} { + for _, settlementErr := range []error{nil, fmt.Errorf("settlement unavailable")} { + t.Run(fmt.Sprintf("final=%v settlementError=%v", final, settlementErr != nil), func(t *testing.T) { + ctrl := gomock.NewController(t) + deliveryChan := make(chan extqueue.Delivery, 1) + mockSub := queuemock.NewMockSubscriber(ctrl) + mockSub.EXPECT().Subscribe(gomock.Any(), "start", gomock.Any()).Return(deliveryChan, nil) + mockQ := queuemock.NewMockQueue(ctrl) + mockQ.EXPECT().Subscriber().Return(mockSub) + config := extqueue.DefaultSubscriptionConfig("worker", "test-group") + if final { + config = extqueue.DLQSubscriptionConfig("worker", "test-group") + } + registry, err := NewTopicRegistry([]TopicConfig{{Key: testTopicKeyStart, Name: "start", Queue: mockQ, Subscription: config}}) + require.NoError(t, err) + c := New(zaptest.NewLogger(t).Sugar(), tally.NoopScope, registry, errs.NewClassifierProcessor(), consumergatenoop.New()) + handler := &testController{name: "test-handler", topicKey: testTopicKeyStart, consumerGroup: "test-group", + processFunc: func(context.Context, Delivery) error { + assert.Fail(t, "controller must not receive inconsistent delivery identity") + return nil + }, + } + require.NoError(t, c.Register(handler)) + require.NoError(t, c.Start(context.Background())) + + msg := entityqueue.NewMessage("mismatch", []byte("payload"), "partition1", map[string]string{entityqueue.MetadataKeyQueueName: "tenant-b"}) + msg.Tenant = "tenant-a" + settled := make(chan struct{}) + delivery := queuemock.NewMockDelivery(ctrl) + delivery.EXPECT().Message().Return(msg).AnyTimes() + settle := func(context.Context, failure.Failure) error { + close(settled) + return settlementErr + } + if final { + delivery.EXPECT().Nack(gomock.Any(), gomock.Any()).DoAndReturn(settle) + } else { + delivery.EXPECT().Reject(gomock.Any(), gomock.Any()).DoAndReturn(settle) + } + deliveryChan <- delivery + <-settled + require.NoError(t, c.Stop(30000)) + }) + } + } } func TestConsumer_ProcessDelivery_RejectsEmptyTenant(t *testing.T) { From aec85e9134523d74e3c39091db37bce17f69264f Mon Sep 17 00:00:00 2001 From: sergeyb Date: Tue, 10 Nov 2026 20:02:13 +0000 Subject: [PATCH 2/2] docs(consumer): place rejection policy at dispatch boundary Summary: Keep the same permanent-input rejection explanation at the pre-dispatch loop boundary. This leaves the original guard body untouched and allows the independent gateway change to land in either order without a historical-hunk conflict. Test Plan: Existing consumer, SQL delivery, and consumer integration targets pass. The net diff versus main remains only the three-line policy comment. Revert Plan: Remove the explanatory comment without changing logged rejection behavior. --- platform/consumer/consumer.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go index 494784a7f..8240f16c3 100644 --- a/platform/consumer/consumer.go +++ b/platform/consumer/consumer.go @@ -289,6 +289,9 @@ func (m *consumer) consumeLoop(ctx context.Context, controller Controller, topic partitionChs := make(map[entityqueue.PartitionIdentity]chan extqueue.Delivery) var wg sync.WaitGroup + // Unchanged invalid metadata is a permanent input error; the failure is logged + // for operator investigation. Without a downstream DLQ, rejection may acknowledge + // and drop the delivery without retry. for { select { case <-ctx.Done(): @@ -350,9 +353,6 @@ func (m *consumer) rejectDeliveryWithInvalidTenant(ctx context.Context, controll "tenant", msg.Tenant, "error", err, ) - // Unchanged invalid metadata is a permanent input error; the failure is logged - // for operator investigation. Without a downstream DLQ, rejection may acknowledge - // and drop the delivery without retry. if rejectErr := delivery.Reject(ctx, failure.New(err.Error())); rejectErr != nil { m.logger.Errorw("failed to reject message with inconsistent queue identity", "controller", controller.Name(),