feat(messagequeue)!: drop the requeue delay parameter from Nack

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)

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",

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
 

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)

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 —

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.

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)

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)

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

platform/extension/messagequeue/mysql/subscriber.go

@@ -238,24 +238,24 @@ 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()
 
 	if d.acknowledged {
 		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
 	}
 
 	d.subscriber.logger.Debugw("message nacked",
 		"topic", d.topic,
 		"partition_key", d.partitionKey,
 		"message_id", d.messageID,
-		"requeue_after_millis", requeueAfterMillis,
 	)
 
 	d.acknowledged = true

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()

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