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

## 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`.

Author: behinddwalls

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/mock/delivery_mock.go

@@ -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/README.md

@@ -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.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)