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