From 18feb7a5124dcf85f5e523bfc43ef3c0a3a28eb0 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Sat, 1 Aug 2026 11:01:06 -0700 Subject: [PATCH] feat(messagequeue)!: drop the requeue delay parameter from Nack MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? Nack's delay parameter was vestigial. The consumer framework — the only production caller — always passed 0 and let the visibility timeout space retries, and the delay's other conceivable use ("check back later" pacing) is exactly what Postpone now expresses with correct retry accounting. Keeping the parameter left two overlapping delay knobs on the delivery contract and invited nack-as-backoff, which burns the DLQ budget. ### What? Delivery.Nack becomes Nack(ctx): the message is immediately eligible for redelivery, retries are spaced by the visibility timeout, and the redelivery counts toward the failure budget as before. MarkNacked drops its delay accordingly (invisible_until = now). Integration tests: TestNackWithDelay is deleted (the delay was the feature; redelivery-after-lapse is already covered by TestVisibilityTimeoutAndRetry); TestNackDoesNotBlockOtherMessages becomes TestInFlightMessageDoesNotBlockOtherMessages, proving the skip property with an un-finalized in-flight head instead of a 30s nack (arrange-first, mirroring the postpone barrier test it contrasts with); other call sites are mechanical. Docs updated (READMEs, sql-queue RFC nack semantics). ## Test Plan ✅ `make test` (83 targets). ✅ `bazel test //test/integration/extension/messagequeue/...` — DLQ at MaxAttempts via immediate nacks, non-blocking in-flight head, multi-consumer-group independence. ✅ `make fmt`, `make mocks`. --- doc/rfc/sql-queue-rfc.md | 8 +- platform/consumer/consumer.go | 4 +- platform/consumer/consumer_test.go | 6 +- platform/extension/messagequeue/README.md | 8 +- platform/extension/messagequeue/delivery.go | 7 +- .../messagequeue/mock/delivery_mock.go | 8 +- .../extension/messagequeue/mysql/README.md | 4 +- .../mysql/delivery_state_store.go | 8 +- .../mysql/delivery_state_store_test.go | 2 +- .../messagequeue/mysql/mock_stores.go | 8 +- .../extension/messagequeue/mysql/stores.go | 5 +- .../messagequeue/mysql/subscriber.go | 8 +- .../messagequeue/mysql/subscriber_test.go | 2 +- .../extension/messagequeue/mysql/README.md | 2 +- .../messagequeue/mysql/queue_test.go | 82 ++++++------------- 15 files changed, 64 insertions(+), 98 deletions(-) diff --git a/doc/rfc/sql-queue-rfc.md b/doc/rfc/sql-queue-rfc.md index 5739aa19..34bd5c84 100644 --- a/doc/rfc/sql-queue-rfc.md +++ b/doc/rfc/sql-queue-rfc.md @@ -245,7 +245,7 @@ DLQ messages are stored in the same `queue_messages` table under a different top **6. Ack** — Set `acked = TRUE` in delivery state. Watermark advancement is deferred to the poll loop for reduced per-ack latency. All operations are idempotent. -**7. Nack** — Set `invisible_until = now + delay` for retry after backoff +**7. Nack** — Set `invisible_until = now`; the message is immediately eligible for redelivery (the visibility timeout is what spaces retries) **8. DLQ** — If `retry_count >= MaxAttempts`: atomically move message to DLQ topic (INSERT with DLQ topic + DELETE from original topic in transaction). MoveToDLQ must succeed before marking acked — otherwise the message would be lost from both main queue and DLQ. @@ -271,13 +271,13 @@ By default, the poll loop fetches a batch of messages (`BatchSize`, default 10) ### Non-Blocking Nack -When a message is nacked, its `invisible_until` is set to a future timestamp. On the next poll, the nacked message is skipped (not deliverable) while subsequent messages are still delivered normally. A nacked message does not block, starve, or delay any other message in the partition. +A nacked message becomes immediately eligible for redelivery; while it is invisible (in flight, or waiting out a visibility lapse), subsequent messages are still delivered normally. A nacked message does not block, starve, or delay any other message in the partition. (A *postponed* message is deliberately the opposite: it acts as a barrier its partition waits behind — see the messagequeue README.) Example with 5 messages at offsets 1-5, all delivered: -- Message 3 is nacked with 30s delay +- Message 3 is nacked - Messages 1, 2, 4, 5 can be acked independently - Watermark advances to 2 (contiguous from head), stops at 3 (not acked) -- After 30s, message 3 becomes deliverable again, is redelivered +- Message 3 is redelivered on a subsequent poll - Once message 3 is acked, watermark jumps from 2 to 5 ### Strict Serialization (Opt-In) diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go index 0126d64c..937363c2 100644 --- a/platform/consumer/consumer.go +++ b/platform/consumer/consumer.go @@ -471,9 +471,9 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d "elapsed_ms", elapsed.Milliseconds(), ) - // Nack with no delay - let visibility timeout handle retry delay + // Nack requeues immediately - the visibility timeout spaces retries nackOp := metrics.Begin(controllerScope, "nack", metrics.StorageLatencyBuckets) - nackErr := delivery.Nack(ctx, 0) + nackErr := delivery.Nack(ctx) nackOp.Complete(nackErr) if nackErr != nil { m.logger.Errorw("failed to nack message", diff --git a/platform/consumer/consumer_test.go b/platform/consumer/consumer_test.go index 4509697f..2a432a19 100644 --- a/platform/consumer/consumer_test.go +++ b/platform/consumer/consumer_test.go @@ -115,7 +115,7 @@ func setupDelivery(del *queuemock.MockDelivery, msg entityqueue.Message, ackErr, close(done) return ackErr }).MaxTimes(1) - del.EXPECT().Nack(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, requeueAfterMillis int64) error { + del.EXPECT().Nack(gomock.Any()).DoAndReturn(func(ctx context.Context) error { close(done) return nackErr }).MaxTimes(1) @@ -458,7 +458,7 @@ func TestConsumer_ProcessDelivery_Hold(t *testing.T) { return tt.postponeErr }) case "nack": - mockDel.EXPECT().Nack(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, requeueAfterMillis int64) error { + mockDel.EXPECT().Nack(gomock.Any()).DoAndReturn(func(ctx context.Context) error { close(done) return nil }) @@ -922,7 +922,7 @@ func TestConsumer_PerPartitionProcessing(t *testing.T) { mockDelA.EXPECT().Metadata().Return(nil).AnyTimes() mockDelA.EXPECT().DeliveryID().Return(msgA.ID).AnyTimes() mockDelA.EXPECT().Ack(gomock.Any()).Return(nil).MaxTimes(1) - mockDelA.EXPECT().Nack(gomock.Any(), gomock.Any()).Return(nil).MaxTimes(1) + mockDelA.EXPECT().Nack(gomock.Any()).Return(nil).MaxTimes(1) deliveryChan <- mockDelA diff --git a/platform/extension/messagequeue/README.md b/platform/extension/messagequeue/README.md index 6f221d78..128adecf 100644 --- a/platform/extension/messagequeue/README.md +++ b/platform/extension/messagequeue/README.md @@ -36,7 +36,7 @@ Message with acknowledgment operations. type Delivery interface { Message() entityqueue.Message Ack(ctx context.Context) error - Nack(ctx context.Context, requeueAfterMillis int64) error + Nack(ctx context.Context) error Postpone(ctx context.Context, delayMs int64) error Reject(ctx context.Context, reason string) error ExtendVisibilityTimeout(ctx context.Context, durationMillis int64) error @@ -48,12 +48,12 @@ type Delivery interface { ``` - **Ack** — message processed successfully, remove from queue -- **Nack** — processing failed, requeue for retry after delay +- **Nack** — processing failed, requeue for immediate retry - **Postpone** — processed successfully but must wait: redeliver after delay, without consuming retry budget; the message is a barrier its partition waits behind - **Reject** — poison pill, move to DLQ (or ack if DLQ disabled) - **ExtendVisibilityTimeout** — extend processing window for long-running work -**`Postpone` vs `Nack` vs `ExtendVisibilityTimeout`:** all three can produce "next delivery happens at T+delay", but they mean different things. `Nack` is a failure — it counts toward `Retry.MaxAttempts` and eventually trips the DLQ, and later offsets in the partition keep flowing past the nacked message (a failed message must not halt its partition). `Postpone` is a deliberate wait — it resets the failure streak (the redelivery restarts at attempt 1) and blocks the partition behind it until it redelivers, in order. `ExtendVisibilityTimeout` is neither: the delivery is still being processed and stays in flight. +**`Postpone` vs `Nack` vs `ExtendVisibilityTimeout`:** `Nack` is a failure — the message is immediately eligible again, the redelivery counts toward `Retry.MaxAttempts` and eventually trips the DLQ, and later offsets in the partition keep flowing past the nacked message (a failed message must not halt its partition). `Postpone` is a deliberate wait — the redelivery happens after the chosen delay, resets the failure streak (it restarts at attempt 1), and blocks the partition behind it until it redelivers, in order. `ExtendVisibilityTimeout` is neither: the delivery is still being processed and stays in flight. ### SubscriptionConfig @@ -89,7 +89,7 @@ cfg := extqueue.DefaultSubscriptionConfig("worker-1", "consumer-group") deliveries, _ := sub.Subscribe(ctx, "topic", cfg) for delivery := range deliveries { if err := process(delivery.Message().Payload); err != nil { - delivery.Nack(ctx, 0) // Retry + delivery.Nack(ctx) // Retry continue } delivery.Ack(ctx) diff --git a/platform/extension/messagequeue/delivery.go b/platform/extension/messagequeue/delivery.go index 9643ad95..9b56a56a 100644 --- a/platform/extension/messagequeue/delivery.go +++ b/platform/extension/messagequeue/delivery.go @@ -36,9 +36,10 @@ type Delivery interface { Ack(ctx context.Context) error // Nack negatively acknowledges the message, indicating processing failure. - // The message will be requeued for redelivery after requeueAfterMillis. - // If requeueAfterMillis is 0, the message is requeued immediately. - Nack(ctx context.Context, requeueAfterMillis int64) error + // The message is requeued for redelivery immediately; the visibility + // timeout is what spaces retries (a crash or missed ack redelivers on the + // same schedule). The redelivery counts toward the failure budget. + Nack(ctx context.Context) error // Postpone finishes this delivery as "processed successfully, redeliver // later": the message becomes invisible for delayMs and acts as a barrier — diff --git a/platform/extension/messagequeue/mock/delivery_mock.go b/platform/extension/messagequeue/mock/delivery_mock.go index 10e9a7c8..e591a747 100644 --- a/platform/extension/messagequeue/mock/delivery_mock.go +++ b/platform/extension/messagequeue/mock/delivery_mock.go @@ -126,17 +126,17 @@ func (mr *MockDeliveryMockRecorder) Metadata() *gomock.Call { } // Nack mocks base method. -func (m *MockDelivery) Nack(ctx context.Context, requeueAfterMillis int64) error { +func (m *MockDelivery) Nack(ctx context.Context) error { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Nack", ctx, requeueAfterMillis) + ret := m.ctrl.Call(m, "Nack", ctx) ret0, _ := ret[0].(error) return ret0 } // Nack indicates an expected call of Nack. -func (mr *MockDeliveryMockRecorder) Nack(ctx, requeueAfterMillis any) *gomock.Call { +func (mr *MockDeliveryMockRecorder) Nack(ctx any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Nack", reflect.TypeOf((*MockDelivery)(nil).Nack), ctx, requeueAfterMillis) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Nack", reflect.TypeOf((*MockDelivery)(nil).Nack), ctx) } // Postpone mocks base method. diff --git a/platform/extension/messagequeue/mysql/README.md b/platform/extension/messagequeue/mysql/README.md index 11ca8bc3..a19765e4 100644 --- a/platform/extension/messagequeue/mysql/README.md +++ b/platform/extension/messagequeue/mysql/README.md @@ -33,7 +33,7 @@ subConfig := extqueue.DefaultSubscriptionConfig("worker-1", "orchestrator") deliveryCh, _ := q.Subscriber().Subscribe(ctx, "merge_events", subConfig) for delivery := range deliveryCh { if err := process(delivery.Message()); err != nil { - delivery.Nack(ctx, 0) // Retry + delivery.Nack(ctx) // Retry continue } delivery.Ack(ctx) @@ -194,4 +194,4 @@ Requires Docker running: bazel test //test/integration/extension/messagequeue/... --test_output=streamed ``` -Integration tests cover: publish/subscribe, partition isolation, ordering, visibility timeout, nack with delay, idempotent publish, concurrent publishers, crash recovery, multiple consumer groups, rebalancing, DLQ, graceful shutdown, non-blocking nack, strict serialization (`BatchSize=1`), and independent consumer group state. +Integration tests cover: publish/subscribe, partition isolation, ordering, visibility timeout, idempotent publish, concurrent publishers, crash recovery, multiple consumer groups, rebalancing, DLQ, graceful shutdown, non-blocking in-flight messages, the postpone barrier, strict serialization (`BatchSize=1`), and independent consumer group state. diff --git a/platform/extension/messagequeue/mysql/delivery_state_store.go b/platform/extension/messagequeue/mysql/delivery_state_store.go index 4d92a793..ff7078ee 100644 --- a/platform/extension/messagequeue/mysql/delivery_state_store.go +++ b/platform/extension/messagequeue/mysql/delivery_state_store.go @@ -149,17 +149,17 @@ func (s *sqldeliveryStateStore) MarkAcked(ctx context.Context, consumerGroup, to return nil } -// MarkNacked sets invisible_until = now + delay to schedule redelivery. +// MarkNacked sets invisible_until = now, making the message immediately +// eligible for redelivery on the next poll. // retry_count is NOT incremented here — it is incremented by MarkDelivered on redelivery. -func (s *sqldeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, delayMs int64) (retErr error) { +func (s *sqldeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) (retErr error) { op := metrics.Begin(s.scope, "mark_nacked", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic), metrics.NewTag("consumer_group", consumerGroup), metrics.NewTag("partition_key", partitionKey)) defer func() { op.Complete(retErr) }() - now := time.Now().UnixMilli() - invisibleUntil := now + delayMs + invisibleUntil := time.Now().UnixMilli() _, err := s.db.ExecContext(ctx, fmt.Sprintf(` INSERT INTO %s (consumer_group, topic, partition_key, message_offset, acked, invisible_until, retry_count) diff --git a/platform/extension/messagequeue/mysql/delivery_state_store_test.go b/platform/extension/messagequeue/mysql/delivery_state_store_test.go index 253c1ab4..89235b8d 100644 --- a/platform/extension/messagequeue/mysql/delivery_state_store_test.go +++ b/platform/extension/messagequeue/mysql/delivery_state_store_test.go @@ -212,7 +212,7 @@ func TestDeliveryStateStore_MarkNacked(t *testing.T) { WillReturnResult(sqlmock.NewResult(1, 1)) } - err := store.MarkNacked(context.Background(), "group-1", "orders", "part-1", 5, 5000) + err := store.MarkNacked(context.Background(), "group-1", "orders", "part-1", 5) if tt.wantErr { require.Error(t, err) diff --git a/platform/extension/messagequeue/mysql/mock_stores.go b/platform/extension/messagequeue/mysql/mock_stores.go index 844c7af3..f8d7d105 100644 --- a/platform/extension/messagequeue/mysql/mock_stores.go +++ b/platform/extension/messagequeue/mysql/mock_stores.go @@ -475,17 +475,17 @@ func (mr *MockdeliveryStateStoreMockRecorder) MarkDelivered(ctx, consumerGroup, } // MarkNacked mocks base method. -func (m *MockdeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset, delayMs int64) error { +func (m *MockdeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) error { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "MarkNacked", ctx, consumerGroup, topic, partitionKey, offset, delayMs) + ret := m.ctrl.Call(m, "MarkNacked", ctx, consumerGroup, topic, partitionKey, offset) ret0, _ := ret[0].(error) return ret0 } // MarkNacked indicates an expected call of MarkNacked. -func (mr *MockdeliveryStateStoreMockRecorder) MarkNacked(ctx, consumerGroup, topic, partitionKey, offset, delayMs any) *gomock.Call { +func (mr *MockdeliveryStateStoreMockRecorder) MarkNacked(ctx, consumerGroup, topic, partitionKey, offset any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MarkNacked", reflect.TypeOf((*MockdeliveryStateStore)(nil).MarkNacked), ctx, consumerGroup, topic, partitionKey, offset, delayMs) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MarkNacked", reflect.TypeOf((*MockdeliveryStateStore)(nil).MarkNacked), ctx, consumerGroup, topic, partitionKey, offset) } // MarkPostponed mocks base method. diff --git a/platform/extension/messagequeue/mysql/stores.go b/platform/extension/messagequeue/mysql/stores.go index 868da242..8c39070f 100644 --- a/platform/extension/messagequeue/mysql/stores.go +++ b/platform/extension/messagequeue/mysql/stores.go @@ -170,8 +170,9 @@ type deliveryStateStore interface { // MarkAcked sets acked = TRUE to indicate this group has processed the message. MarkAcked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) error - // MarkNacked sets invisible_until = now + delay to schedule redelivery. - MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, delayMs int64) error + // MarkNacked makes the message immediately eligible for redelivery + // (invisible_until = now). + MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) error // MarkPostponed sets invisible_until = now + delay, resets retry_count, and // sets the postponed flag. The message becomes a partition barrier until it diff --git a/platform/extension/messagequeue/mysql/subscriber.go b/platform/extension/messagequeue/mysql/subscriber.go index 5d5fd377..a63b289e 100644 --- a/platform/extension/messagequeue/mysql/subscriber.go +++ b/platform/extension/messagequeue/mysql/subscriber.go @@ -238,7 +238,7 @@ func (d *sqlDelivery) Ack(ctx context.Context) error { } // Nack implements extqueue.Delivery.Nack -func (d *sqlDelivery) Nack(ctx context.Context, requeueAfterMillis int64) error { +func (d *sqlDelivery) Nack(ctx context.Context) error { d.mu.Lock() defer d.mu.Unlock() @@ -246,8 +246,9 @@ func (d *sqlDelivery) Nack(ctx context.Context, requeueAfterMillis int64) error return &ErrAlreadyAcknowledged{DeliveryID: d.deliveryID} } - // Mark as nacked in delivery state (per consumer group, with delay and retry_count) - if err := d.subscriber.deliveryStateStore.MarkNacked(ctx, d.consumerGroup, d.topic, d.partitionKey, d.offset, requeueAfterMillis); err != nil { + // Mark as nacked in delivery state (per consumer group): immediately + // eligible for redelivery on the next poll. + if err := d.subscriber.deliveryStateStore.MarkNacked(ctx, d.consumerGroup, d.topic, d.partitionKey, d.offset); err != nil { return err } @@ -255,7 +256,6 @@ func (d *sqlDelivery) Nack(ctx context.Context, requeueAfterMillis int64) error "topic", d.topic, "partition_key", d.partitionKey, "message_id", d.messageID, - "requeue_after_millis", requeueAfterMillis, ) d.acknowledged = true diff --git a/platform/extension/messagequeue/mysql/subscriber_test.go b/platform/extension/messagequeue/mysql/subscriber_test.go index d5c12cb5..c021397a 100644 --- a/platform/extension/messagequeue/mysql/subscriber_test.go +++ b/platform/extension/messagequeue/mysql/subscriber_test.go @@ -49,7 +49,7 @@ func newTestDeliveryStateStore(ctrl *gomock.Controller) *MockdeliveryStateStore mockDS := NewMockdeliveryStateStore(ctrl) mockDS.EXPECT().MarkDelivered(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(0, nil).AnyTimes() mockDS.EXPECT().MarkAcked(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() - mockDS.EXPECT().MarkNacked(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + mockDS.EXPECT().MarkNacked(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() mockDS.EXPECT().GetDeliveryState(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(DeliveryState{}, false, nil).AnyTimes() mockDS.EXPECT().AdvanceWatermark(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(int64(0), nil).AnyTimes() mockDS.EXPECT().ExtendVisibility(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() diff --git a/test/integration/extension/messagequeue/mysql/README.md b/test/integration/extension/messagequeue/mysql/README.md index 08b6654b..a340116b 100644 --- a/test/integration/extension/messagequeue/mysql/README.md +++ b/test/integration/extension/messagequeue/mysql/README.md @@ -49,7 +49,7 @@ Signal names describe behavioral concerns, not implementation details, so they r ## Test categories - **Publish/subscribe basics** — ordering, metadata, partitioning, late subscribers, idempotency -- **Visibility and retry** — timeout expiry, `ExtendVisibilityTimeout`, nack with delay +- **Visibility and retry** — timeout expiry, `ExtendVisibilityTimeout`, nack redelivery - **Crash recovery** — worker crash with in-flight messages, reject + crash, retry-limit + crash - **Consumer groups** — independent state, multiple workers in a group, load balancing - **Rebalance** — even distribution, subscriber leave, odd partitions, excess subscribers diff --git a/test/integration/extension/messagequeue/mysql/queue_test.go b/test/integration/extension/messagequeue/mysql/queue_test.go index e4c35b10..fd913905 100644 --- a/test/integration/extension/messagequeue/mysql/queue_test.go +++ b/test/integration/extension/messagequeue/mysql/queue_test.go @@ -630,46 +630,6 @@ func (s *SQLQueueIntegrationSuite) TestVisibilityTimeoutAndRetry() { t.Logf("Successfully tested ExtendVisibilityTimeout and visibility timeout retry") } -func (s *SQLQueueIntegrationSuite) TestNackWithDelay() { - t := s.T() - - q, err := queueMySQL.NewQueue(queueMySQL.Params{ - DB: s.db, - Logger: zaptest.NewLogger(t), - MetricsScope: tally.NoopScope, - }) - require.NoError(t, err) - defer q.Close() - - publisher := q.Publisher() - subscriber := q.Subscriber() - - topic := "nack_topic" - - // Subscribe - subConfig := testSubConfig("worker-1", "nack-consumer") - deliveryChan, err := subscriber.Subscribe(s.ctx, topic, subConfig) - require.NoError(t, err) - - // Publish message - msg := entityqueue.NewMessage("nack-msg", []byte("test"), "nack-partition", nil) - require.NoError(t, publisher.Publish(s.ctx, topic, msg)) - - // Receive and Nack with delay - nackDelay := 2 * time.Second - - delivery := receive(t, deliveryChan) - t.Logf("Received message, nacking with %s delay", nackDelay) - nackErr := delivery.Nack(s.ctx, int64(nackDelay.Milliseconds())) - require.NoError(t, nackErr) - - // Should receive again after nack delay — subscriber polls and finds msg visible - delivery2 := receive(t, deliveryChan) - t.Logf("Received message again after nack delay") - assert.Equal(t, "nack-msg", delivery2.Message().ID) - require.NoError(t, delivery2.Ack(s.ctx)) -} - func (s *SQLQueueIntegrationSuite) TestIdempotentPublish() { t := s.T() @@ -1103,7 +1063,7 @@ func (s *SQLQueueIntegrationSuite) TestDeadLetterQueue() { assert.Equal(t, "poison-msg", delivery.Message().ID) // Nack without delay to retry immediately - require.NoError(t, delivery.Nack(s.ctx, 0)) + require.NoError(t, delivery.Nack(s.ctx)) } // After MaxAttempts, message should be moved to DLQ topic @@ -2010,7 +1970,7 @@ func (s *SQLQueueIntegrationSuite) TestRebalance_MoreSubscribersThanPartitions() // TestNackDoesNotBlockOtherMessages verifies that nacking a message does not // block delivery of subsequent messages in the same partition. The nacked // message should be skipped (invisible) while later messages are delivered. -func (s *SQLQueueIntegrationSuite) TestNackDoesNotBlockOtherMessages() { +func (s *SQLQueueIntegrationSuite) TestInFlightMessageDoesNotBlockOtherMessages() { t := s.T() q, err := queueMySQL.NewQueue(queueMySQL.Params{ @@ -2022,26 +1982,29 @@ func (s *SQLQueueIntegrationSuite) TestNackDoesNotBlockOtherMessages() { topic := "nack_nonblocking_topic" partition := "nack-nb-part" - // Subscribe with batch=10 to fetch multiple messages per poll + // Subscribe with batch=10 to fetch multiple messages per poll. The default + // 60s visibility timeout keeps msg-1 invisible for the whole test. subConfig := extqueue.DefaultSubscriptionConfig("worker-1", "nack-nb-cg") subConfig.PollIntervalMs = 50 subConfig.BatchSize = 10 deliveryCh, err := q.Subscriber().Subscribe(s.ctx, topic, subConfig) require.NoError(t, err) - // Publish 3 messages in order - for i := 1; i <= 3; i++ { + // Publish the first message alone and receive it, leaving it in flight + // (un-finalized, invisible) at the lowest offset of the partition. + msg1 := entityqueue.NewMessage("msg-1", []byte("payload-1"), partition, nil) + require.NoError(t, q.Publisher().Publish(s.ctx, topic, msg1)) + d1 := receive(t, deliveryCh) + assert.Equal(t, "msg-1", d1.Message().ID) + t.Logf("msg-1 in flight (invisible), publishing later offsets") + + // Later offsets must still be deliverable despite the invisible msg-1 — + // the opposite of a postponed message, which is a barrier. + for i := 2; i <= 3; i++ { msg := entityqueue.NewMessage(fmt.Sprintf("msg-%d", i), []byte(fmt.Sprintf("payload-%d", i)), partition, nil) require.NoError(t, q.Publisher().Publish(s.ctx, topic, msg)) } - // Receive first message and nack it with a long delay - d1 := receive(t, deliveryCh) - assert.Equal(t, "msg-1", d1.Message().ID) - require.NoError(t, d1.Nack(s.ctx, 30000)) // 30s delay — won't come back during test - t.Logf("Nacked msg-1 with 30s delay") - - // Messages 2 and 3 should still be deliverable despite msg-1 being nacked d2 := receive(t, deliveryCh) assert.Equal(t, "msg-2", d2.Message().ID) require.NoError(t, d2.Ack(s.ctx)) @@ -2052,7 +2015,8 @@ func (s *SQLQueueIntegrationSuite) TestNackDoesNotBlockOtherMessages() { require.NoError(t, d3.Ack(s.ctx)) t.Logf("Received and acked msg-3") - t.Logf("Verified: nacked message did not block subsequent messages") + require.NoError(t, d1.Ack(s.ctx)) + t.Logf("Verified: an in-flight (invisible) message did not block subsequent messages") } // TestPostponeBlocksPartitionUntilDue verifies the postpone barrier: while a @@ -2103,7 +2067,7 @@ func (s *SQLQueueIntegrationSuite) TestPostponeBlocksPartitionUntilDue() { } // Barrier: messages 2 and 3 must not be delivered while msg-1 waits — - // the opposite of the nacked case above. + // the opposite of the in-flight case above. assertNoDelivery(t, deliveryCh, signalCh, queueMySQL.SignalDeliveryCheck, 3) t.Logf("Confirmed: partition blocked behind postponed msg-1") @@ -2165,7 +2129,7 @@ func (s *SQLQueueIntegrationSuite) TestPostponeResetsRetryBudget() { delivery := receive(t, deliveryChan) assert.Equal(t, attempt, delivery.Attempt()) assert.Equal(t, "wait-then-poison", delivery.Message().ID) - require.NoError(t, delivery.Nack(s.ctx, 0)) + require.NoError(t, delivery.Nack(s.ctx)) t.Logf("Attempt %d: nacked", delivery.Attempt()) } @@ -2270,7 +2234,7 @@ func (s *SQLQueueIntegrationSuite) TestMultipleConsumerGroupsIndependentState() // CG-alpha: nack msg-1, ack msg-2 d1a := receive(t, ch1) assert.Equal(t, "shared-1", d1a.Message().ID) - require.NoError(t, d1a.Nack(s.ctx, 200)) // short nack delay + require.NoError(t, d1a.Nack(s.ctx)) t.Logf("cg-alpha nacked shared-1") d2a := receive(t, ch1) @@ -2454,14 +2418,14 @@ func (s *SQLQueueIntegrationSuite) TestCrashAfterRetryLimitDoesNotLoseMessages() require.NoError(t, deliveries["msg-A"].Ack(s.ctx)) t.Logf("Acked msg-A") - // Nack B with short delay so it becomes visible quickly for redelivery - require.NoError(t, deliveries["msg-B"].Nack(s.ctx, 100)) + // Nack B — immediately visible again for redelivery + require.NoError(t, deliveries["msg-B"].Nack(s.ctx)) t.Logf("Nacked msg-B, waiting for retry-limit to trigger auto-DLQ") // Do NOT ack msg-C — simulating in-flight at crash time. // Wait for msg-B to be redelivered and auto-DLQ'd by the poll loop. - // The poll loop picks up msg-B after 100ms nack delay, sees retry_count >= MaxAttempts, moves to DLQ. + // The poll loop picks up the nacked msg-B, sees retry_count >= MaxAttempts, moves it to DLQ. // We just need to wait long enough for that to happen before crashing. // A short sleep is acceptable here as we're waiting for the subscriber's // internal processing, not for a test condition. But let's use receive