feat(platform): hold/postpone — a fourth delivery outcome for backing off (#487)

## Summary
### Why?

Queue controllers have no way to say "this message is fine, but it must
wait." The outcome model is ack/nack/reject, so waiting stages ack and
republish fresh copies of their own messages via PublishAfter, with
message-id minting to dodge the publish dedup, retry accounting reset
every cycle, and loop liveness hanging on a publish succeeding. Designed
in doc/rfc/consumer-hold.md (previous commit).

### What?

The messagequeue extension Delivery gains Postpone(delayMs): the
delivery finalizes, the message becomes invisible for the delay, and it
acts as a partition barrier — the mysql poll loop stops scanning the
partition at a postponed row instead of skipping past it (nacked rows
keep skip-and-continue, so failures never halt a partition). A new
`postponed` flag on queue_delivery_state makes the post-postpone
redelivery exempt from the retry_count increment and resets the count,
so deliberate waits never burn the DLQ budget while real failures still
dead-letter.

The consumer framework's Delivery view gains Hold(delayMs): an
intent-recording call with no I/O. On a nil return from Process the
framework postpones instead of acking (metric op `postpone`); an error
return wins over a recorded hold (`hold_ignored` counter). A failed
postpone write is abandoned like a failed ack — the visibility timeout
lapses into a normal redelivery, so hold-loop liveness is
framework-owned.

Controller unit tests across submitqueue/stovepipe/runway previously
passed the extension mock as consumer.Delivery, which only worked
structurally; they now use the consumer-facing mock (which has Hold).

## Test Plan
✅ `make test` (83 targets) — includes new consumer hold outcome tests,
mysql MarkPostponed/GetDeliveryState store tests, and poll-loop barrier
tests. ✅ `bazel test //test/integration/extension/messagequeue/...` —
new end-to-end tests: postpone blocks the partition until due then
redelivers in order as attempt 1; a postpone resets the budget but
subsequent real failures still dead-letter. ✅ `make fmt`, `make
gazelle`, `make mocks`.

## Issues


## Stack
1. #486
1. @ #487
1. #488
1. #489
1. #490
1. #491
1. #492

Author: behinddwalls

platform/consumer/README.md

@@ -62,7 +62,7 @@ type Controller interface {
 
 ### Delivery
 
-A restricted view of a queue delivery exposed to controllers. Hides Ack/Nack/Reject (handled automatically by Consumer) while exposing message data and `ExtendVisibilityTimeout`.
+A restricted view of a queue delivery exposed to controllers. Hides Ack/Nack/Reject (handled automatically by Consumer) while exposing message data, `ExtendVisibilityTimeout`, and `Hold`.
 
 ## TopicRegistry
 
@@ -92,6 +92,7 @@ registry, _ := consumer.NewTopicRegistry([]consumer.TopicConfig{
 The consumer passes every non-nil controller error through the configured `errs.ErrorProcessor` once and then uses `errs.IsRetryable` to decide the transport action:
 
 - **`return nil`** — success, message is acked.
+- **`delivery.Hold(delayMs)` then `return nil`** — success that chose to wait: the message is postponed instead of acked. It redelivers after the delay as a barrier its partition waits behind, and the redelivery does not count toward the retry limit (`Attempt()` restarts at 1). A hold is only honored on success — if `Process` returns an error, the failure outcome below wins and the recorded hold is discarded (logged, `hold_ignored` counter). Use hold for backoff loops (waiting for a budget slot, polling an external status) instead of acking and republishing to your own topic.
 - **non-nil, retryable after processing** — message is nacked for redelivery (visibility timeout drives the retry delay).
 - **non-nil, non-retryable after processing** — message is rejected, which moves it to the DLQ if one is configured for the subscription, or simply acks-and-drops if not.
 
@@ -120,7 +121,21 @@ When the consumer is wired with `errs.AlwaysRetryableProcessor` (DLQ reconciliat
 
 The consumer records controller operations with `process.start` and `process.finish`. The finish histogram records both latency and completion count with `result=success|error|cancel`; error and cancellation series also include `origin=infra|infra_retryable|user` and `dependency=yes|no`. These dimensions are added after error processing, so they describe the classified error that drives ack, nack, or reject behavior rather than the controller's raw return value. The lifecycle histogram count replaces separate received, processed, and controller-error counters.
 
-The consumer also owns lifecycle metrics for the resulting `ack`, `nack`, or `reject` transport operation. Queue controllers should emit only domain-specific event counters; they must not duplicate the consumer-owned `process` lifecycle metrics.
+The consumer also owns lifecycle metrics for the resulting `ack`, `nack`, `postpone`, or `reject` transport operation. Queue controllers should emit only domain-specific event counters; they must not duplicate the consumer-owned `process` lifecycle metrics.
+
+## Which wait do I want?
+
+Several mechanisms can delay work; they mean different things. Pick by what you're trying to say:
+
+| You want to say | Use | Partition while waiting |
+|---|---|---|
+| "This delivery failed — retry it" | return a retryable error (framework nacks) | keeps flowing — a failure never halts its partition |
+| "I'm still working — keep my lease" | `delivery.ExtendVisibilityTimeout(...)` | blocked behind the in-flight delivery |
+| "Done for now — wake this partition in N ms" | `delivery.Hold(N)` then `return nil` | paused behind the postponed message (barrier), redelivers first in order |
+| "Stop this controller/partition from outside" (tests, operators) | consumer gate (`platform/extension/consumergate`) | parked in flight until the gate opens |
+| "Defer *other* work" — a delayed message to another topic or key | `Publisher.PublishAfter` | not involved — it's a fresh publish |
+
+Gate vs hold, since both pause a partition: the **gate** is an external, event-ended stop — someone stops the controller at the door, before `Process` ever sees the message. **Hold** is a controller-chosen, timer-ended wait — the controller saw the work and decided to come back later. Business logic never closes or opens gates; a controller that needs to back off uses hold.
 
 ## Lifecycle
 

platform/consumer/consumer.go

@@ -411,6 +411,18 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 	op.Complete(err, completionTags...)
 
 	if err != nil {
+		// A failure outcome wins over a recorded hold — a hold is only honored
+		// on success, so retry accounting and dead-lettering stay meaningful.
+		if wrapped.held {
+			metrics.NamedCounter(controllerScope, opName, "hold_ignored", 1)
+			m.logger.Warnw("hold recorded but controller returned error, failure outcome wins",
+				"controller", controller.Name(),
+				"topic_key", topicKey,
+				"message_id", msg.ID,
+				"partition_key", msg.PartitionKey,
+			)
+		}
+
 		// By convention, Controller can only return context.Canceled if it is
 		// cancelled by the processing context during shutdown.
 		isCanceled := errors.Is(err, context.Canceled)
@@ -474,6 +486,36 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 		return
 	}
 
+	// Controller succeeded with a recorded hold - postpone instead of acking.
+	// The message redelivers after the delay as a partition barrier, without
+	// consuming retry budget. A failed postpone is abandoned like a failed ack:
+	// the visibility timeout lapses into a normal redelivery, so the hold
+	// loop's liveness never depends on this write succeeding.
+	if wrapped.held {
+		postponeOp := metrics.Begin(controllerScope, "postpone", metrics.StorageLatencyBuckets)
+		postponeErr := delivery.Postpone(ctx, wrapped.holdDelayMs)
+		postponeOp.Complete(postponeErr)
+		if postponeErr != nil {
+			m.logger.Errorw("failed to postpone held message",
+				"controller", controller.Name(),
+				"topic_key", topicKey,
+				"message_id", msg.ID,
+				"error", postponeErr,
+			)
+			return
+		}
+
+		m.logger.Debugw("message held, postponed for redelivery",
+			"controller", controller.Name(),
+			"topic_key", topicKey,
+			"message_id", msg.ID,
+			"partition_key", msg.PartitionKey,
+			"delay_ms", wrapped.holdDelayMs,
+			"elapsed_ms", elapsed.Milliseconds(),
+		)
+		return
+	}
+
 	// Controller succeeded - ack message
 	ackOp := metrics.Begin(controllerScope, "ack", metrics.StorageLatencyBuckets)
 	ackErr := delivery.Ack(ctx)

platform/consumer/consumer_test.go

@@ -357,6 +357,125 @@ func TestConsumer_ProcessDelivery_Error(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestConsumer_ProcessDelivery_Hold(t *testing.T) {
+	tests := []struct {
+		name        string
+		processFunc func(ctx context.Context, delivery Delivery) error
+		postponeErr error
+		// wantOutcome is the delivery method the framework must call: "postpone" or "nack".
+		wantOutcome string
+		wantDelayMs int64
+	}{
+		{
+			name: "hold postpones instead of acking",
+			processFunc: func(ctx context.Context, delivery Delivery) error {
+				delivery.Hold(5000)
+				return nil
+			},
+			wantOutcome: "postpone",
+			wantDelayMs: 5000,
+		},
+		{
+			name: "last hold wins",
+			processFunc: func(ctx context.Context, delivery Delivery) error {
+				delivery.Hold(1000)
+				delivery.Hold(2500)
+				return nil
+			},
+			wantOutcome: "postpone",
+			wantDelayMs: 2500,
+		},
+		{
+			name: "negative delay clamps to zero",
+			processFunc: func(ctx context.Context, delivery Delivery) error {
+				delivery.Hold(-5)
+				return nil
+			},
+			wantOutcome: "postpone",
+			wantDelayMs: 0,
+		},
+		{
+			name: "error outcome wins over hold",
+			processFunc: func(ctx context.Context, delivery Delivery) error {
+				delivery.Hold(5000)
+				return errs.NewRetryableError(fmt.Errorf("processing failed"))
+			},
+			wantOutcome: "nack",
+		},
+		{
+			name: "postpone failure leaves delivery in flight",
+			processFunc: func(ctx context.Context, delivery Delivery) error {
+				delivery.Hold(3000)
+				return nil
+			},
+			postponeErr: fmt.Errorf("db error"),
+			wantOutcome: "postpone",
+			wantDelayMs: 3000,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			logger := zaptest.NewLogger(t).Sugar()
+
+			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)
+
+			reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group")
+
+			c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
+
+			handler := &testController{}
+			setupController(handler, "test-handler", testTopicKeyStart, "test-group", tt.processFunc)
+
+			require.NoError(t, c.Register(handler))
+
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			require.NoError(t, c.Start(ctx))
+
+			msg := entityqueue.NewMessage("held-msg", []byte("payload"), "partition1", nil)
+			done := make(chan struct{})
+			var gotDelayMs int64
+			mockDel := queuemock.NewMockDelivery(ctrl)
+			mockDel.EXPECT().Message().Return(msg).AnyTimes()
+			mockDel.EXPECT().Attempt().Return(1).AnyTimes()
+			mockDel.EXPECT().ReceivedAt().Return(time.Now().UnixMilli()).AnyTimes()
+			mockDel.EXPECT().Metadata().Return(nil).AnyTimes()
+			mockDel.EXPECT().DeliveryID().Return(msg.ID).AnyTimes()
+			// No Ack expectation: an Ack call on a held delivery fails the test.
+			switch tt.wantOutcome {
+			case "postpone":
+				mockDel.EXPECT().Postpone(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, delayMs int64) error {
+					gotDelayMs = delayMs
+					close(done)
+					return tt.postponeErr
+				})
+			case "nack":
+				mockDel.EXPECT().Nack(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, requeueAfterMillis int64) error {
+					close(done)
+					return nil
+				})
+			}
+
+			deliveryChan <- mockDel
+			<-done
+
+			if tt.wantOutcome == "postpone" {
+				assert.Equal(t, tt.wantDelayMs, gotDelayMs)
+			}
+
+			require.NoError(t, c.Stop(30000))
+		})
+	}
+}
+
 func TestConsumer_ProcessDelivery_NonRetryableError(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	logger := zaptest.NewLogger(t).Sugar()

platform/consumer/controller.go

@@ -26,13 +26,14 @@ import (
 // Delivery is the consumer package's view of a queue delivery.
 // It exists to hide Ack/Nack from controllers — the Consumer framework handles those
 // automatically based on the error returned from Process(). Controllers only see
-// message data, metadata, and ExtendVisibilityTimeout (a business-level concern for
-// long-running processing).
+// message data, metadata, ExtendVisibilityTimeout (a business-level concern for
+// long-running processing), and Hold (a business-level concern for backing off).
 //
 // To signal outcome from Process():
 //   - Return nil to ack the message (success).
 //   - Return an error to nack the message for retry.
 //   - Return a non-retryable error to reject a poison pill message (removes it from the queue).
+//   - Call Hold(delayMs) and return nil to postpone the message (redeliver later, partition waits).
 type Delivery interface {
 	// Message returns the delivered message.
 	Message() entityqueue.Message
@@ -41,11 +42,21 @@ type Delivery interface {
 	// visible to other consumers. Use when processing takes longer than expected.
 	ExtendVisibilityTimeout(ctx context.Context, durationMillis int64) error
 
+	// Hold records intent to postpone this delivery: when Process then returns
+	// nil, the framework postpones the message for delayMs instead of acking.
+	// The postponed message is a barrier — its partition is not consumed past
+	// it until it redelivers, in order — and the redelivery does not count
+	// toward the retry limit. Recording has no side effects; the last call
+	// wins; a negative delay is clamped to 0. If Process returns an error, the
+	// failure outcome wins and the recorded hold is discarded. Must be called
+	// from the Process goroutine before returning.
+	Hold(delayMs int64)
+
 	// DeliveryID returns a backend-specific identifier for this delivery.
 	DeliveryID() string
 
 	// Attempt returns how many times this message has been delivered.
-	// Starts at 1 for first delivery.
+	// Starts at 1 for first delivery. A postponed redelivery restarts at 1.
 	Attempt() int
 
 	// ReceivedAt returns when this delivery was received (Unix milliseconds).
@@ -59,6 +70,11 @@ type Delivery interface {
 // Hides Ack/Nack from controllers - Consumer handles those automatically.
 type deliveryWrapper struct {
 	delivery extqueue.Delivery
+
+	// held and holdDelayMs record Hold intent. Written from the Process
+	// goroutine, read by the framework after Process returns.
+	held        bool
+	holdDelayMs int64
 }
 
 func (d *deliveryWrapper) Message() entityqueue.Message {
@@ -69,6 +85,14 @@ func (d *deliveryWrapper) ExtendVisibilityTimeout(ctx context.Context, durationM
 	return d.delivery.ExtendVisibilityTimeout(ctx, durationMillis)
 }
 
+func (d *deliveryWrapper) Hold(delayMs int64) {
+	if delayMs < 0 {
+		delayMs = 0
+	}
+	d.held = true
+	d.holdDelayMs = delayMs
+}
+
 func (d *deliveryWrapper) DeliveryID() string {
 	return d.delivery.DeliveryID()
 }
@@ -96,6 +120,7 @@ type Controller interface {
 	// Process processes a delivery. Controller receives consumer.Delivery (not extension/entityqueue.Delivery)
 	// which prevents direct Ack/Nack calls - Consumer handles those automatically.
 	// Return nil to ack the message (success), error to nack and retry, or NonRetryableError to ack a poison pill message.
+	// Call delivery.Hold(delayMs) and return nil to postpone the message instead of acking it.
 	// Context controls the lifecycle of the service. It is cancelled when the consumer is stopped. The implementation should process it gracefully:
 	//  - Pass the context to the underlying services and wait for them to complete their operations.
 	//  - Proceed to the nearest safe state.

platform/consumer/mock/controller_mock.go

@@ -25,7 +25,7 @@ type Publisher interface {
 - `Nack` is "this delivery failed, try again" — it bumps `retry_count` and eventually trips DLQ.
 - `PublishAfter` is "postpone this work" — `retry_count` resets to 0, DLQ stays available for true failures.
 
-Use `PublishAfter` for self-driven poll loops (e.g. the orchestrator's `buildsignal` consumer re-publishing itself between `Status` calls). Use `Nack` for processing failures.
+For a consumer deferring its *own current delivery* ("check back in N ms"), prefer `Delivery.Postpone` (below) over ack-plus-`PublishAfter`: it needs no publisher, no fresh message id, and keeps the same log row. `PublishAfter` remains the tool for deferring *other* work — publishing a delayed message to a different topic or key. Use `Nack` for processing failures.
 
 ### Subscriber
 Consumes messages from topics with per-subscription configuration.
@@ -45,6 +45,7 @@ type Delivery interface {
     Message() entityqueue.Message
     Ack(ctx context.Context) error
     Nack(ctx context.Context, requeueAfterMillis int64) error
+    Postpone(ctx context.Context, delayMs int64) error
     Reject(ctx context.Context, reason string) error
     ExtendVisibilityTimeout(ctx context.Context, durationMillis int64) error
     DeliveryID() string
@@ -56,9 +57,12 @@ type Delivery interface {
 
 - **Ack** — message processed successfully, remove from queue
 - **Nack** — processing failed, requeue for retry after delay
+- **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.
+
 ### SubscriptionConfig
 
 Per-subscription configuration for polling, batching, leasing, retries, and DLQ:

platform/extension/messagequeue/README.md

@@ -40,6 +40,14 @@ type Delivery interface {
 	// If requeueAfterMillis is 0, the message is requeued immediately.
 	Nack(ctx context.Context, requeueAfterMillis int64) error
 
+	// Postpone finishes this delivery as "processed successfully, redeliver
+	// later": the message becomes invisible for delayMs and acts as a barrier —
+	// its partition is not consumed past it until it redelivers, in order.
+	// Unlike Nack, the redelivery does not count against the failure budget
+	// (retry limit / DLQ); postponing resets the failure streak.
+	// Postpone is terminal for this delivery, like Ack/Nack/Reject.
+	Postpone(ctx context.Context, delayMs int64) error
+
 	// Reject moves the message to the dead letter entityqueue.
 	// Use for poison pill messages that should never be retried.
 	// reason is stored as last_error in the DLQ for debugging.