From 770b37644baf01ce29dd3d2eeeeb1ad0455b7602 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Sat, 1 Aug 2026 09:46:52 -0700 Subject: [PATCH] =?UTF-8?q?feat(platform):=20hold/postpone=20=E2=80=94=20a?= =?UTF-8?q?=20fourth=20delivery=20outcome=20for=20backing=20off?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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`. --- platform/consumer/README.md | 19 +- platform/consumer/consumer.go | 42 ++++ platform/consumer/consumer_test.go | 119 ++++++++++++ platform/consumer/controller.go | 31 ++- platform/consumer/mock/controller_mock.go | 12 ++ platform/extension/messagequeue/README.md | 6 +- platform/extension/messagequeue/delivery.go | 8 + .../messagequeue/mock/delivery_mock.go | 14 ++ .../extension/messagequeue/mysql/README.md | 4 +- .../mysql/delivery_state_store.go | 48 ++++- .../mysql/delivery_state_store_test.go | 64 +++++- .../messagequeue/mysql/mock_stores.go | 14 ++ .../mysql/schema/queue_delivery_state.sql | 8 +- .../extension/messagequeue/mysql/stores.go | 12 +- .../messagequeue/mysql/subscriber.go | 34 +++- .../messagequeue/mysql/subscriber_test.go | 182 ++++++++++++++++++ platform/extension/messagequeue/publisher.go | 9 +- runway/controller/dlq/BUILD.bazel | 1 + runway/controller/dlq/dlq_test.go | 5 +- runway/controller/merge/BUILD.bazel | 1 + runway/controller/merge/merge_test.go | 5 +- .../controller/mergeconflictcheck/BUILD.bazel | 1 + .../mergeconflictcheck_test.go | 5 +- stovepipe/controller/build/BUILD.bazel | 1 + stovepipe/controller/build/build_test.go | 3 +- stovepipe/controller/buildsignal/BUILD.bazel | 1 + .../buildsignal/buildsignal_test.go | 3 +- stovepipe/controller/dlq/BUILD.bazel | 2 +- stovepipe/controller/dlq/dlq_test.go | 4 +- stovepipe/controller/process/BUILD.bazel | 1 + stovepipe/controller/process/process_test.go | 3 +- .../gateway/controller/log/BUILD.bazel | 2 +- .../gateway/controller/log/log_test.go | 4 +- .../orchestrator/controller/batch/BUILD.bazel | 1 + .../controller/batch/batch_test.go | 25 +-- .../orchestrator/controller/build/BUILD.bazel | 1 + .../controller/build/build_test.go | 15 +- .../controller/buildsignal/BUILD.bazel | 1 + .../buildsignal/buildsignal_test.go | 5 +- .../controller/cancel/BUILD.bazel | 1 + .../controller/cancel/cancel_test.go | 3 +- .../controller/conclude/BUILD.bazel | 1 + .../controller/conclude/conclude_test.go | 5 +- .../orchestrator/controller/dlq/BUILD.bazel | 1 + .../controller/dlq/request_test.go | 6 +- .../orchestrator/controller/merge/BUILD.bazel | 1 + .../controller/merge/merge_test.go | 5 +- .../mergeconflictsignal/BUILD.bazel | 1 + .../mergeconflictsignal_test.go | 5 +- .../controller/mergesignal/BUILD.bazel | 1 + .../mergesignal/mergesignal_test.go | 5 +- .../controller/speculate/BUILD.bazel | 1 + .../controller/speculate/speculate_test.go | 5 +- .../orchestrator/controller/start/BUILD.bazel | 1 + .../controller/start/start_test.go | 7 +- .../controller/validate/BUILD.bazel | 1 + .../controller/validate/validate_test.go | 23 +-- .../messagequeue/mysql/queue_test.go | 129 +++++++++++++ 58 files changed, 829 insertions(+), 89 deletions(-) diff --git a/platform/consumer/README.md b/platform/consumer/README.md index 6bd067db..8c0507ff 100644 --- a/platform/consumer/README.md +++ b/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 diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go index 01fb6a63..0126d64c 100644 --- a/platform/consumer/consumer.go +++ b/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) diff --git a/platform/consumer/consumer_test.go b/platform/consumer/consumer_test.go index 3e0331da..4509697f 100644 --- a/platform/consumer/consumer_test.go +++ b/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() diff --git a/platform/consumer/controller.go b/platform/consumer/controller.go index 0b31b05c..ec335b8a 100644 --- a/platform/consumer/controller.go +++ b/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. diff --git a/platform/consumer/mock/controller_mock.go b/platform/consumer/mock/controller_mock.go index ebb2d82a..29aae494 100644 --- a/platform/consumer/mock/controller_mock.go +++ b/platform/consumer/mock/controller_mock.go @@ -84,6 +84,18 @@ func (mr *MockDeliveryMockRecorder) ExtendVisibilityTimeout(ctx, durationMillis return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ExtendVisibilityTimeout", reflect.TypeOf((*MockDelivery)(nil).ExtendVisibilityTimeout), ctx, durationMillis) } +// Hold mocks base method. +func (m *MockDelivery) Hold(delayMs int64) { + m.ctrl.T.Helper() + m.ctrl.Call(m, "Hold", delayMs) +} + +// Hold indicates an expected call of Hold. +func (mr *MockDeliveryMockRecorder) Hold(delayMs any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Hold", reflect.TypeOf((*MockDelivery)(nil).Hold), delayMs) +} + // Message mocks base method. func (m *MockDelivery) Message() messagequeue.Message { m.ctrl.T.Helper() diff --git a/platform/extension/messagequeue/README.md b/platform/extension/messagequeue/README.md index 1f39ec56..c3bc9b2b 100644 --- a/platform/extension/messagequeue/README.md +++ b/platform/extension/messagequeue/README.md @@ -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: diff --git a/platform/extension/messagequeue/delivery.go b/platform/extension/messagequeue/delivery.go index 4a180c70..9643ad95 100644 --- a/platform/extension/messagequeue/delivery.go +++ b/platform/extension/messagequeue/delivery.go @@ -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. diff --git a/platform/extension/messagequeue/mock/delivery_mock.go b/platform/extension/messagequeue/mock/delivery_mock.go index 7aa8bd24..10e9a7c8 100644 --- a/platform/extension/messagequeue/mock/delivery_mock.go +++ b/platform/extension/messagequeue/mock/delivery_mock.go @@ -139,6 +139,20 @@ func (mr *MockDeliveryMockRecorder) Nack(ctx, requeueAfterMillis any) *gomock.Ca return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Nack", reflect.TypeOf((*MockDelivery)(nil).Nack), ctx, requeueAfterMillis) } +// Postpone mocks base method. +func (m *MockDelivery) Postpone(ctx context.Context, delayMs int64) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Postpone", ctx, delayMs) + ret0, _ := ret[0].(error) + return ret0 +} + +// Postpone indicates an expected call of Postpone. +func (mr *MockDeliveryMockRecorder) Postpone(ctx, delayMs any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Postpone", reflect.TypeOf((*MockDelivery)(nil).Postpone), ctx, delayMs) +} + // ReceivedAt mocks base method. func (m *MockDelivery) ReceivedAt() int64 { m.ctrl.T.Helper() diff --git a/platform/extension/messagequeue/mysql/README.md b/platform/extension/messagequeue/mysql/README.md index 38671b15..9c8d7709 100644 --- a/platform/extension/messagequeue/mysql/README.md +++ b/platform/extension/messagequeue/mysql/README.md @@ -107,13 +107,15 @@ platform/extension/messagequeue/mysql/ | Table | Purpose | Scoped To | |-------|---------|-----------| | `queue_messages` | Immutable append-only message log | `(topic, partition_key)` — shared across consumer groups | -| `queue_delivery_state` | Visibility, ack state, retry count | `(consumer_group, topic, partition_key, offset)` | +| `queue_delivery_state` | Visibility, ack state, retry count, postponed flag | `(consumer_group, topic, partition_key, offset)` | | `queue_offsets` | Contiguous acked watermark | `(consumer_group, topic, partition_key)` | | `queue_partition_leases` | Partition lease coordination | `(consumer_group, topic, partition_key)` | | `queue_subscriber_heartbeats` | Active subscriber tracking | `(consumer_group, topic, subscriber_name)` | `queue_messages` has a `visible_after BIGINT UNSIGNED NOT NULL DEFAULT 0` column that supports `Publisher.PublishAfter`: subscribers' `FetchByOffset` skips rows where `visible_after > now`. Default 0 means immediately visible, so existing rows continue to behave as before — the column is back-compatible. +`queue_delivery_state` has a `postponed BOOLEAN NOT NULL DEFAULT FALSE` column that supports `Delivery.Postpone`. `MarkPostponed` sets `invisible_until = now + delay`, resets `retry_count` to 0, and sets the flag. While the flag is set and the row is invisible, the poll loop treats the message as a **barrier** — it stops scanning the partition instead of skipping past it (nacked rows keep skip-and-continue semantics, so a failed message never halts its partition). On the next `MarkDelivered` the flag is consumed: the `retry_count` increment is skipped and the flag cleared, so a postponed redelivery restarts as attempt 1 and only consecutive real failures count toward `Retry.MaxAttempts`. Default FALSE keeps existing rows back-compatible. + See `schema/` for full SQL definitions. See the [RFC](../../../doc/rfc/sql-queue-rfc.md#database-schema) for field-level documentation. ### Store Architecture diff --git a/platform/extension/messagequeue/mysql/delivery_state_store.go b/platform/extension/messagequeue/mysql/delivery_state_store.go index 9828b476..4d92a793 100644 --- a/platform/extension/messagequeue/mysql/delivery_state_store.go +++ b/platform/extension/messagequeue/mysql/delivery_state_store.go @@ -58,12 +58,17 @@ func (s *sqldeliveryStateStore) MarkDelivered(ctx context.Context, consumerGroup now := time.Now().UnixMilli() invisibleUntil := now + visibilityTimeoutMs + // Assignment order matters: retry_count reads the pre-update postponed value + // (MySQL applies ON DUPLICATE KEY UPDATE assignments left to right), so the + // postponed reset must come after it. A postponed redelivery is a deliberate + // wait, not a failure — it is exempt from the increment and consumes the flag. _, err := s.db.ExecContext(ctx, fmt.Sprintf(` - INSERT INTO %s (consumer_group, topic, partition_key, message_offset, acked, invisible_until, retry_count) - VALUES (?, ?, ?, ?, FALSE, ?, 0) + INSERT INTO %s (consumer_group, topic, partition_key, message_offset, acked, invisible_until, retry_count, postponed) + VALUES (?, ?, ?, ?, FALSE, ?, 0, FALSE) ON DUPLICATE KEY UPDATE invisible_until = IF(acked = FALSE, VALUES(invisible_until), invisible_until), - retry_count = IF(acked = FALSE, retry_count + 1, retry_count) + retry_count = IF(acked = FALSE AND postponed = FALSE, retry_count + 1, retry_count), + postponed = IF(acked = FALSE, FALSE, postponed) `, DeliveryStateTableName), consumerGroup, topic, partitionKey, offset, invisibleUntil) @@ -171,6 +176,39 @@ func (s *sqldeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, t return nil } +// MarkPostponed sets invisible_until = now + delay, resets retry_count, and sets +// the postponed flag. A postpone is a deliberate wait, not a failure: the flag +// makes the message a partition barrier while invisible, and exempts the next +// MarkDelivered from the retry_count increment. The reset restarts failure +// accounting — a completed delivery that chose to wait has demonstrated the +// message is processable. +func (s *sqldeliveryStateStore) MarkPostponed(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, delayMs int64) (retErr error) { + op := metrics.Begin(s.scope, "mark_postponed", 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 + + _, err := s.db.ExecContext(ctx, fmt.Sprintf(` + INSERT INTO %s (consumer_group, topic, partition_key, message_offset, acked, invisible_until, retry_count, postponed) + VALUES (?, ?, ?, ?, FALSE, ?, 0, TRUE) + ON DUPLICATE KEY UPDATE + invisible_until = IF(acked = FALSE, VALUES(invisible_until), invisible_until), + retry_count = IF(acked = FALSE, 0, retry_count), + postponed = IF(acked = FALSE, TRUE, postponed) + `, DeliveryStateTableName), + consumerGroup, topic, partitionKey, offset, invisibleUntil) + + if err != nil { + return fmt.Errorf("mark postponed topic=%s partition=%s offset=%d: %w", topic, partitionKey, offset, err) + } + + return nil +} + // GetDeliveryState returns the full delivery state for a message offset. // Returns (state, found, error). found=false means no row (never delivered). func (s *sqldeliveryStateStore) GetDeliveryState(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) (_ DeliveryState, _ bool, retErr error) { @@ -182,9 +220,9 @@ func (s *sqldeliveryStateStore) GetDeliveryState(ctx context.Context, consumerGr var state DeliveryState err := s.db.QueryRowContext(ctx, fmt.Sprintf(` - SELECT acked, invisible_until, retry_count FROM %s + SELECT acked, invisible_until, retry_count, postponed FROM %s WHERE consumer_group = ? AND topic = ? AND partition_key = ? AND message_offset = ? - `, DeliveryStateTableName), consumerGroup, topic, partitionKey, offset).Scan(&state.Acked, &state.InvisibleUntil, &state.RetryCount) + `, DeliveryStateTableName), consumerGroup, topic, partitionKey, offset).Scan(&state.Acked, &state.InvisibleUntil, &state.RetryCount, &state.Postponed) if err == sql.ErrNoRows { return DeliveryState{}, false, nil diff --git a/platform/extension/messagequeue/mysql/delivery_state_store_test.go b/platform/extension/messagequeue/mysql/delivery_state_store_test.go index 0b49b08a..253c1ab4 100644 --- a/platform/extension/messagequeue/mysql/delivery_state_store_test.go +++ b/platform/extension/messagequeue/mysql/delivery_state_store_test.go @@ -224,12 +224,55 @@ func TestDeliveryStateStore_MarkNacked(t *testing.T) { } } +func TestDeliveryStateStore_MarkPostponed(t *testing.T) { + tests := []struct { + name string + wantErr bool + }{ + { + name: "success", + wantErr: false, + }, + { + name: "db error", + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store, db, mock := newTestDeliveryStateStoreWithMock(t) + defer db.Close() + + if tt.wantErr { + mock.ExpectExec("INSERT INTO queue_delivery_state"). + WithArgs("group-1", "orders", "part-1", int64(5), sqlmock.AnyArg()). + WillReturnError(assert.AnError) + } else { + mock.ExpectExec("INSERT INTO queue_delivery_state"). + WithArgs("group-1", "orders", "part-1", int64(5), sqlmock.AnyArg()). + WillReturnResult(sqlmock.NewResult(1, 1)) + } + + err := store.MarkPostponed(context.Background(), "group-1", "orders", "part-1", 5, 5000) + + if tt.wantErr { + require.Error(t, err) + } else { + require.NoError(t, err) + } + assert.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + func TestDeliveryStateStore_GetDeliveryState(t *testing.T) { tests := []struct { name string acked bool invisibleUntil int64 retryCount int + postponed bool noRows bool wantErr bool wantFound bool @@ -260,6 +303,14 @@ func TestDeliveryStateStore_GetDeliveryState(t *testing.T) { retryCount: 3, wantFound: true, }, + { + name: "postponed message", + acked: false, + invisibleUntil: 9999999999999, + retryCount: 0, + postponed: true, + wantFound: true, + }, { name: "db error", wantErr: true, @@ -272,18 +323,18 @@ func TestDeliveryStateStore_GetDeliveryState(t *testing.T) { defer db.Close() if tt.wantErr { - mock.ExpectQuery("SELECT acked, invisible_until, retry_count FROM queue_delivery_state"). + mock.ExpectQuery("SELECT acked, invisible_until, retry_count, postponed FROM queue_delivery_state"). WithArgs("group-1", "orders", "part-1", int64(5)). WillReturnError(assert.AnError) } else if tt.noRows { - mock.ExpectQuery("SELECT acked, invisible_until, retry_count FROM queue_delivery_state"). + mock.ExpectQuery("SELECT acked, invisible_until, retry_count, postponed FROM queue_delivery_state"). WithArgs("group-1", "orders", "part-1", int64(5)). - WillReturnRows(sqlmock.NewRows([]string{"acked", "invisible_until", "retry_count"})) + WillReturnRows(sqlmock.NewRows([]string{"acked", "invisible_until", "retry_count", "postponed"})) } else { - mock.ExpectQuery("SELECT acked, invisible_until, retry_count FROM queue_delivery_state"). + mock.ExpectQuery("SELECT acked, invisible_until, retry_count, postponed FROM queue_delivery_state"). WithArgs("group-1", "orders", "part-1", int64(5)). - WillReturnRows(sqlmock.NewRows([]string{"acked", "invisible_until", "retry_count"}). - AddRow(tt.acked, tt.invisibleUntil, tt.retryCount)) + WillReturnRows(sqlmock.NewRows([]string{"acked", "invisible_until", "retry_count", "postponed"}). + AddRow(tt.acked, tt.invisibleUntil, tt.retryCount, tt.postponed)) } state, found, err := store.GetDeliveryState(context.Background(), "group-1", "orders", "part-1", 5) @@ -297,6 +348,7 @@ func TestDeliveryStateStore_GetDeliveryState(t *testing.T) { assert.Equal(t, tt.acked, state.Acked) assert.Equal(t, tt.invisibleUntil, state.InvisibleUntil) assert.Equal(t, tt.retryCount, state.RetryCount) + assert.Equal(t, tt.postponed, state.Postponed) } } assert.NoError(t, mock.ExpectationsWereMet()) diff --git a/platform/extension/messagequeue/mysql/mock_stores.go b/platform/extension/messagequeue/mysql/mock_stores.go index f5bff7f5..87af009c 100644 --- a/platform/extension/messagequeue/mysql/mock_stores.go +++ b/platform/extension/messagequeue/mysql/mock_stores.go @@ -501,3 +501,17 @@ func (mr *MockdeliveryStateStoreMockRecorder) MarkNacked(ctx, consumerGroup, top mr.mock.ctrl.T.Helper() return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MarkNacked", reflect.TypeOf((*MockdeliveryStateStore)(nil).MarkNacked), ctx, consumerGroup, topic, partitionKey, offset, delayMs) } + +// MarkPostponed mocks base method. +func (m *MockdeliveryStateStore) MarkPostponed(ctx context.Context, consumerGroup, topic, partitionKey string, offset, delayMs int64) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "MarkPostponed", ctx, consumerGroup, topic, partitionKey, offset, delayMs) + ret0, _ := ret[0].(error) + return ret0 +} + +// MarkPostponed indicates an expected call of MarkPostponed. +func (mr *MockdeliveryStateStoreMockRecorder) MarkPostponed(ctx, consumerGroup, topic, partitionKey, offset, delayMs any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MarkPostponed", reflect.TypeOf((*MockdeliveryStateStore)(nil).MarkPostponed), ctx, consumerGroup, topic, partitionKey, offset, delayMs) +} diff --git a/platform/extension/messagequeue/mysql/schema/queue_delivery_state.sql b/platform/extension/messagequeue/mysql/schema/queue_delivery_state.sql index 1b9ddc8f..27625cdc 100644 --- a/platform/extension/messagequeue/mysql/schema/queue_delivery_state.sql +++ b/platform/extension/messagequeue/mysql/schema/queue_delivery_state.sql @@ -4,7 +4,7 @@ -- -- State encoding: -- acked = TRUE → processed, never redeliver --- acked = FALSE, invisible_until > now → in-flight or nack delay +-- acked = FALSE, invisible_until > now → in-flight, nack delay, or postpone delay -- acked = FALSE, invisible_until <= now → ready for (re-)delivery CREATE TABLE IF NOT EXISTS queue_delivery_state ( @@ -31,5 +31,11 @@ CREATE TABLE IF NOT EXISTS queue_delivery_state ( -- Number of times this message has been redelivered to this consumer group retry_count INT UNSIGNED NOT NULL DEFAULT 0, + -- Whether the last delivery was postponed (deliberate wait, not a failure). + -- While set and invisible, the message is a barrier: its partition is not + -- consumed past it. The next delivery is exempt from the retry_count + -- increment and clears the flag. + postponed BOOLEAN NOT NULL DEFAULT FALSE, + PRIMARY KEY (consumer_group, topic, partition_key, message_offset) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; diff --git a/platform/extension/messagequeue/mysql/stores.go b/platform/extension/messagequeue/mysql/stores.go index 01a60fb4..424c5b8f 100644 --- a/platform/extension/messagequeue/mysql/stores.go +++ b/platform/extension/messagequeue/mysql/stores.go @@ -155,12 +155,17 @@ type DeliveryState struct { InvisibleUntil int64 // RetryCount tracks how many times the message has been delivered RetryCount int + // Postponed indicates the last delivery was postponed (a deliberate wait, + // not a failure). While set and invisible, the message is a partition + // barrier and its next delivery is exempt from the retry_count increment. + Postponed bool } // deliveryStateStore handles per-consumer-group delivery tracking (internal use only) type deliveryStateStore interface { // MarkDelivered inserts a row marking message as in-flight for this consumer group. - // Increments retry_count on redelivery (ON DUPLICATE KEY UPDATE). + // Increments retry_count on redelivery (ON DUPLICATE KEY UPDATE), except when the + // row is marked postponed — that delivery is exempt and clears the postponed flag. // Returns the resulting retry_count after the operation. MarkDelivered(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, visibilityTimeoutMs int64) (retryCount int, err error) @@ -174,6 +179,11 @@ type deliveryStateStore interface { // MarkNacked sets invisible_until = now + delay to schedule redelivery. MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, delayMs int64) error + // MarkPostponed sets invisible_until = now + delay, resets retry_count, and + // sets the postponed flag. The message becomes a partition barrier until it + // redelivers, and the redelivery does not count as a failure. + MarkPostponed(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, delayMs int64) error + // GetDeliveryState returns the full delivery state for a message offset. // Returns (state, found, error). found=false means no row (never delivered). GetDeliveryState(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) (DeliveryState, bool, error) diff --git a/platform/extension/messagequeue/mysql/subscriber.go b/platform/extension/messagequeue/mysql/subscriber.go index ffc80532..62a5769c 100644 --- a/platform/extension/messagequeue/mysql/subscriber.go +++ b/platform/extension/messagequeue/mysql/subscriber.go @@ -262,6 +262,32 @@ func (d *sqlDelivery) Nack(ctx context.Context, requeueAfterMillis int64) error return nil } +// Postpone implements extqueue.Delivery.Postpone +func (d *sqlDelivery) Postpone(ctx context.Context, delayMs int64) error { + d.mu.Lock() + defer d.mu.Unlock() + + if d.acknowledged { + return &ErrAlreadyAcknowledged{DeliveryID: d.deliveryID} + } + + // Mark as postponed in delivery state (per consumer group): invisible for + // the delay, retry_count reset, partition barrier until redelivery. + if err := d.subscriber.deliveryStateStore.MarkPostponed(ctx, d.consumerGroup, d.topic, d.partitionKey, d.offset, delayMs); err != nil { + return err + } + + d.subscriber.logger.Debugw("message postponed", + "topic", d.topic, + "partition_key", d.partitionKey, + "message_id", d.messageID, + "delay_millis", delayMs, + ) + + d.acknowledged = true + return nil +} + // Reject implements extqueue.Delivery.Reject func (d *sqlDelivery) Reject(ctx context.Context, reason string) error { d.mu.Lock() @@ -792,9 +818,15 @@ func (w *partitionWorker) pollAndDeliver(ctx context.Context) (retErr error) { // Determine deliverability in-memory: // !found → new message, deliverable // state.Acked → already processed, skip - // state.InvisibleUntil > now → in-flight or nack delay, skip + // state.InvisibleUntil > now → in-flight, nack delay, or postpone delay now := time.Now().UnixMilli() if found && (state.Acked || state.InvisibleUntil > now) { + // A postponed message is a barrier: its partition waits for it, so + // stop scanning instead of skipping past it. In-flight and nacked + // messages are skipped — a failed delivery must not halt its partition. + if !state.Acked && state.Postponed { + break + } continue } diff --git a/platform/extension/messagequeue/mysql/subscriber_test.go b/platform/extension/messagequeue/mysql/subscriber_test.go index d5ab8c3f..ea770783 100644 --- a/platform/extension/messagequeue/mysql/subscriber_test.go +++ b/platform/extension/messagequeue/mysql/subscriber_test.go @@ -192,6 +192,77 @@ func TestSQLDelivery_Ack(t *testing.T) { } } +func TestSQLDelivery_Postpone(t *testing.T) { + tests := []struct { + name string + alreadyAcked bool + markPostponedErr error + expectErr bool + }{ + { + name: "successful postpone", + }, + { + name: "already acknowledged returns error", + alreadyAcked: true, + expectErr: true, + }, + { + name: "MarkPostponed failure returns error", + markPostponedErr: fmt.Errorf("db error"), + expectErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockMsgStore := NewMockmessageStore(ctrl) + mockOffStore := NewMockoffsetStore(ctrl) + mockLeaseStore := NewMockpartitionLeaseStore(ctrl) + mockDeliveryState := NewMockdeliveryStateStore(ctrl) + + sub := NewSubscriber( + zaptest.NewLogger(t).Sugar(), + tally.NoopScope, + mockMsgStore, + mockOffStore, + mockLeaseStore, + newTestHeartbeatStore(ctrl), + mockDeliveryState, + ) + + msg := entityqueue.NewMessage("msg-1", []byte("payload"), "part-1", nil) + d := newSQLDelivery( + msg, "1", 1, nil, + sub, "test_topic", "part-1", 100, "msg-1", "test-group", + extqueue.DLQConfig{}, + ) + + if tt.alreadyAcked { + d.acknowledged = true + } + + if !tt.alreadyAcked { + mockDeliveryState.EXPECT().MarkPostponed( + gomock.Any(), "test-group", "test_topic", "part-1", int64(100), int64(5000), + ).Return(tt.markPostponedErr) + } + + err := d.Postpone(context.Background(), 5000) + + if tt.expectErr { + require.Error(t, err) + } else { + require.NoError(t, err) + assert.True(t, d.acknowledged) + } + }) + } +} + func TestSQLDelivery_Reject(t *testing.T) { tests := []struct { name string @@ -550,6 +621,117 @@ func TestSubscriber_PartitionWorkerPollAndDeliver(t *testing.T) { assert.True(t, foundFinish, "expected poll.finish histogram") } +// TestSubscriber_PollAndDeliver_PostponedBarrier verifies that a postponed +// message halts the partition scan (barrier), while a nacked message is +// skipped and later offsets keep flowing. +func TestSubscriber_PollAndDeliver_PostponedBarrier(t *testing.T) { + tests := []struct { + name string + // state of the first row (offset 1); rows 2 and 3 have no delivery state + firstRowState DeliveryState + // expectDeliveries is how many of the later rows are delivered + expectDeliveries int + }{ + { + name: "postponed row is a barrier, later offsets wait", + firstRowState: DeliveryState{ + Acked: false, + InvisibleUntil: time.Now().UnixMilli() + 60000, + Postponed: true, + }, + expectDeliveries: 0, + }, + { + name: "nacked row is skipped, later offsets flow", + firstRowState: DeliveryState{ + Acked: false, + InvisibleUntil: time.Now().UnixMilli() + 60000, + Postponed: false, + }, + expectDeliveries: 2, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockMessageStore := NewMockmessageStore(ctrl) + mockOffsetStore := NewMockoffsetStore(ctrl) + mockLeaseStore := NewMockpartitionLeaseStore(ctrl) + mockDeliveryState := NewMockdeliveryStateStore(ctrl) + + s := NewSubscriber( + zaptest.NewLogger(t).Sugar(), + tally.NoopScope, + mockMessageStore, + mockOffsetStore, + mockLeaseStore, + newTestHeartbeatStore(ctrl), + mockDeliveryState, + ) + + cfg := testSubscriptionConfig() + deliveryCh := make(chan extqueue.Delivery, 10) + sub := &subscription{ + topic: "test_topic", + config: cfg, + deliveryCh: deliveryCh, + workers: make(map[string]*partitionWorker), + } + + ctx := context.Background() + + mockOffsetStore.EXPECT().Initialize(gomock.Any(), "test_topic", "part-1", cfg.ConsumerGroup).Return(nil) + mockOffsetStore.EXPECT().GetAckedOffset(gomock.Any(), "test_topic", "part-1", cfg.ConsumerGroup).Return(int64(0), nil).Times(2) + + rows := []messageRow{ + {ID: "msg-1", Offset: 1, PartitionKey: "part-1", Payload: []byte("p1"), PublishedAt: time.Now().UnixMilli()}, + {ID: "msg-2", Offset: 2, PartitionKey: "part-1", Payload: []byte("p2"), PublishedAt: time.Now().UnixMilli()}, + {ID: "msg-3", Offset: 3, PartitionKey: "part-1", Payload: []byte("p3"), PublishedAt: time.Now().UnixMilli()}, + } + mockMessageStore.EXPECT().FetchByOffset(gomock.Any(), "test_topic", "part-1", int64(0), gomock.Any(), cfg.BatchSize). + Return(rows, nil) + + mockDeliveryState.EXPECT().GetDeliveryState(gomock.Any(), cfg.ConsumerGroup, "test_topic", "part-1", int64(1)). + Return(tt.firstRowState, true, nil) + if tt.expectDeliveries > 0 { + for _, offset := range []int64{2, 3} { + mockDeliveryState.EXPECT().GetDeliveryState(gomock.Any(), cfg.ConsumerGroup, "test_topic", "part-1", offset). + Return(DeliveryState{}, false, nil) + mockDeliveryState.EXPECT().MarkDelivered(gomock.Any(), cfg.ConsumerGroup, "test_topic", "part-1", offset, cfg.VisibilityTimeoutMs). + Return(0, nil) + } + } + + mockMessageStore.EXPECT().GetOffsetsAbove(gomock.Any(), "test_topic", "part-1", int64(0), watermarkAdvancementLimit).Return(nil, nil) + mockDeliveryState.EXPECT().AdvanceWatermark(gomock.Any(), cfg.ConsumerGroup, "test_topic", "part-1", int64(0), gomock.Nil()).Return(int64(0), nil) + + w := &partitionWorker{ + partitionKey: "part-1", + sub: sub, + subscriber: s, + done: make(chan struct{}), + } + + require.NoError(t, w.pollAndDeliver(ctx)) + + delivered := 0 + for { + select { + case <-deliveryCh: + delivered++ + continue + default: + } + break + } + assert.Equal(t, tt.expectDeliveries, delivered) + }) + } +} + // TestSubscriber_StopAllWorkers tests that all workers are stopped gracefully. func TestSubscriber_StopAllWorkers(t *testing.T) { ctrl := gomock.NewController(t) diff --git a/platform/extension/messagequeue/publisher.go b/platform/extension/messagequeue/publisher.go index cddba7bc..0a9f8034 100644 --- a/platform/extension/messagequeue/publisher.go +++ b/platform/extension/messagequeue/publisher.go @@ -33,10 +33,11 @@ type Publisher interface { // it does not consume a delivery_state retry slot. delayMs <= 0 is // equivalent to Publish. // - // Use for "postpone this work" semantics (e.g. spacing out repeated - // poll cycles for a single key). Use Nack with a delay for "this - // delivery failed, try again" — the two signals stay separate so - // retry_count and DLQ behaviour remain meaningful. + // Use for deferring *other* work — a delayed message to another topic or + // key. A consumer deferring its own current delivery should use + // Delivery.Postpone instead; use Nack with a delay for "this delivery + // failed, try again" — the signals stay separate so retry_count and DLQ + // behaviour remain meaningful. PublishAfter(ctx context.Context, topic string, message entityqueue.Message, delayMs int64) error // Close gracefully shuts down the publisher, flushing pending messages. diff --git a/runway/controller/dlq/BUILD.bazel b/runway/controller/dlq/BUILD.bazel index d3b6d5a0..052b95b9 100644 --- a/runway/controller/dlq/BUILD.bazel +++ b/runway/controller/dlq/BUILD.bazel @@ -25,6 +25,7 @@ go_test( "//api/runway/messagequeue/protopb:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", diff --git a/runway/controller/dlq/dlq_test.go b/runway/controller/dlq/dlq_test.go index 42f3a5c5..36085283 100644 --- a/runway/controller/dlq/dlq_test.go +++ b/runway/controller/dlq/dlq_test.go @@ -25,6 +25,7 @@ import ( runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "go.uber.org/mock/gomock" "go.uber.org/zap/zaptest" @@ -42,10 +43,10 @@ type publishedMsg struct { msg entityqueue.Message } -func newDelivery(t *testing.T, ctrl *gomock.Controller, payload []byte, meta map[string]string) *queuemock.MockDelivery { +func newDelivery(t *testing.T, ctrl *gomock.Controller, payload []byte, meta map[string]string) *consumermock.MockDelivery { t.Helper() msg := entityqueue.NewMessage(testID, payload, testPartitionKey, nil) - d := queuemock.NewMockDelivery(ctrl) + d := consumermock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(msg).AnyTimes() d.EXPECT().Metadata().Return(meta).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() diff --git a/runway/controller/merge/BUILD.bazel b/runway/controller/merge/BUILD.bazel index 5a2bf7d7..baa9ad86 100644 --- a/runway/controller/merge/BUILD.bazel +++ b/runway/controller/merge/BUILD.bazel @@ -26,6 +26,7 @@ go_test( "//api/runway/messagequeue/protopb:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//runway/extension/merger:go_default_library", "//runway/extension/merger/mock:go_default_library", diff --git a/runway/controller/merge/merge_test.go b/runway/controller/merge/merge_test.go index bab87891..c7900863 100644 --- a/runway/controller/merge/merge_test.go +++ b/runway/controller/merge/merge_test.go @@ -26,6 +26,7 @@ import ( runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/runway/extension/merger" mergermock "github.com/uber/submitqueue/runway/extension/merger/mock" @@ -39,10 +40,10 @@ const ( testPartitionKey = "test-queue" ) -func newDelivery(t *testing.T, ctrl *gomock.Controller, payload []byte) *queuemock.MockDelivery { +func newDelivery(t *testing.T, ctrl *gomock.Controller, payload []byte) *consumermock.MockDelivery { t.Helper() msg := entityqueue.NewMessage(testID, payload, testPartitionKey, nil) - d := queuemock.NewMockDelivery(ctrl) + d := consumermock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(msg).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() return d diff --git a/runway/controller/mergeconflictcheck/BUILD.bazel b/runway/controller/mergeconflictcheck/BUILD.bazel index a5cefefb..624e9ff9 100644 --- a/runway/controller/mergeconflictcheck/BUILD.bazel +++ b/runway/controller/mergeconflictcheck/BUILD.bazel @@ -26,6 +26,7 @@ go_test( "//api/runway/messagequeue/protopb:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//runway/extension/merger:go_default_library", "//runway/extension/merger/mock:go_default_library", diff --git a/runway/controller/mergeconflictcheck/mergeconflictcheck_test.go b/runway/controller/mergeconflictcheck/mergeconflictcheck_test.go index 31d9d497..13215239 100644 --- a/runway/controller/mergeconflictcheck/mergeconflictcheck_test.go +++ b/runway/controller/mergeconflictcheck/mergeconflictcheck_test.go @@ -26,6 +26,7 @@ import ( runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/runway/extension/merger" mergermock "github.com/uber/submitqueue/runway/extension/merger/mock" @@ -39,10 +40,10 @@ const ( testPartitionKey = "test-queue" ) -func newDelivery(t *testing.T, ctrl *gomock.Controller, payload []byte) *queuemock.MockDelivery { +func newDelivery(t *testing.T, ctrl *gomock.Controller, payload []byte) *consumermock.MockDelivery { t.Helper() msg := entityqueue.NewMessage(testID, payload, testPartitionKey, nil) - d := queuemock.NewMockDelivery(ctrl) + d := consumermock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(msg).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() return d diff --git a/stovepipe/controller/build/BUILD.bazel b/stovepipe/controller/build/BUILD.bazel index bb5dd113..522236a6 100644 --- a/stovepipe/controller/build/BUILD.bazel +++ b/stovepipe/controller/build/BUILD.bazel @@ -27,6 +27,7 @@ go_test( deps = [ "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//stovepipe/core/messagequeue:go_default_library", diff --git a/stovepipe/controller/build/build_test.go b/stovepipe/controller/build/build_test.go index e2aead4f..ebcbf8d9 100644 --- a/stovepipe/controller/build/build_test.go +++ b/stovepipe/controller/build/build_test.go @@ -24,6 +24,7 @@ import ( "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/platform/errs" mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" @@ -82,7 +83,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildMoc func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) consumer.Delivery { t.Helper() - d := mqmock.NewMockDelivery(ctrl) + d := consumermock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(entityqueue.NewMessage(testID, payload, testID, nil)).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() return d diff --git a/stovepipe/controller/buildsignal/BUILD.bazel b/stovepipe/controller/buildsignal/BUILD.bazel index 767d30da..d5b9c17c 100644 --- a/stovepipe/controller/buildsignal/BUILD.bazel +++ b/stovepipe/controller/buildsignal/BUILD.bazel @@ -26,6 +26,7 @@ go_test( deps = [ "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//stovepipe/core/messagequeue:go_default_library", diff --git a/stovepipe/controller/buildsignal/buildsignal_test.go b/stovepipe/controller/buildsignal/buildsignal_test.go index 23e25875..7a2aab48 100644 --- a/stovepipe/controller/buildsignal/buildsignal_test.go +++ b/stovepipe/controller/buildsignal/buildsignal_test.go @@ -24,6 +24,7 @@ import ( "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/platform/errs" mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" @@ -85,7 +86,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildsig func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) consumer.Delivery { t.Helper() - d := mqmock.NewMockDelivery(ctrl) + d := consumermock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(entityqueue.NewMessage(testBuildID, payload, testBuildID, nil)).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() return d diff --git a/stovepipe/controller/dlq/BUILD.bazel b/stovepipe/controller/dlq/BUILD.bazel index 22054560..4e8f713b 100644 --- a/stovepipe/controller/dlq/BUILD.bazel +++ b/stovepipe/controller/dlq/BUILD.bazel @@ -26,7 +26,7 @@ go_test( deps = [ "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", - "//platform/extension/messagequeue/mock:go_default_library", + "//platform/consumer/mock:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/storage:go_default_library", diff --git a/stovepipe/controller/dlq/dlq_test.go b/stovepipe/controller/dlq/dlq_test.go index 06b659de..010ca40a 100644 --- a/stovepipe/controller/dlq/dlq_test.go +++ b/stovepipe/controller/dlq/dlq_test.go @@ -23,7 +23,7 @@ import ( "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" - queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" @@ -60,7 +60,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, dlqMocks func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) consumer.Delivery { t.Helper() - d := queuemock.NewMockDelivery(ctrl) + d := consumermock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(entityqueue.NewMessage(testID, payload, testQueue, nil)).AnyTimes() d.EXPECT().Attempt().Return(4).AnyTimes() d.EXPECT().Metadata().Return(map[string]string{ diff --git a/stovepipe/controller/process/BUILD.bazel b/stovepipe/controller/process/BUILD.bazel index 907112ba..ff45ee27 100644 --- a/stovepipe/controller/process/BUILD.bazel +++ b/stovepipe/controller/process/BUILD.bazel @@ -28,6 +28,7 @@ go_test( deps = [ "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//stovepipe/core/messagequeue:go_default_library", diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index 1ebc8504..1f60edf5 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -24,6 +24,7 @@ import ( "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/platform/errs" mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" @@ -102,7 +103,7 @@ func newControllerWithScope(t *testing.T, ctrl *gomock.Controller, scope tally.S func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) consumer.Delivery { t.Helper() - d := mqmock.NewMockDelivery(ctrl) + d := consumermock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(entityqueue.NewMessage(testID, payload, testQueue, nil)).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() return d diff --git a/submitqueue/gateway/controller/log/BUILD.bazel b/submitqueue/gateway/controller/log/BUILD.bazel index 646a5c1b..eb1743a8 100644 --- a/submitqueue/gateway/controller/log/BUILD.bazel +++ b/submitqueue/gateway/controller/log/BUILD.bazel @@ -22,7 +22,7 @@ go_test( embed = [":go_default_library"], deps = [ "//platform/base/messagequeue:go_default_library", - "//platform/extension/messagequeue/mock:go_default_library", + "//platform/consumer/mock:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/storage/mock:go_default_library", diff --git a/submitqueue/gateway/controller/log/log_test.go b/submitqueue/gateway/controller/log/log_test.go index d5c4019c..2d8a2a37 100644 --- a/submitqueue/gateway/controller/log/log_test.go +++ b/submitqueue/gateway/controller/log/log_test.go @@ -22,7 +22,7 @@ import ( "github.com/stretchr/testify/require" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" - queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" @@ -96,7 +96,7 @@ func TestController_Process(t *testing.T) { } controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, tt.setupStore(ctrl), topickey.TopicKeyLog, "gateway-log") msg := entityqueue.NewMessage("test-queue/1", payload, "test-queue", nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() diff --git a/submitqueue/orchestrator/controller/batch/BUILD.bazel b/submitqueue/orchestrator/controller/batch/BUILD.bazel index 8ded51d9..9394c33d 100644 --- a/submitqueue/orchestrator/controller/batch/BUILD.bazel +++ b/submitqueue/orchestrator/controller/batch/BUILD.bazel @@ -29,6 +29,7 @@ go_test( "//platform/base/mergestrategy:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/extension/counter/mock:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//submitqueue/core/topickey:go_default_library", diff --git a/submitqueue/orchestrator/controller/batch/batch_test.go b/submitqueue/orchestrator/controller/batch/batch_test.go index fdfef8e0..f755d514 100644 --- a/submitqueue/orchestrator/controller/batch/batch_test.go +++ b/submitqueue/orchestrator/controller/batch/batch_test.go @@ -28,6 +28,7 @@ import ( "github.com/uber/submitqueue/platform/base/mergestrategy" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" countermock "github.com/uber/submitqueue/platform/extension/counter/mock" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/submitqueue/core/topickey" @@ -149,7 +150,7 @@ func TestController_Process_Success(t *testing.T) { request := testRequest() msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -211,7 +212,7 @@ func TestController_Process_PublishesBatchedLog(t *testing.T) { ) msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -238,7 +239,7 @@ func TestController_Process_StorageFailure(t *testing.T) { controller := newTestController(t, ctrl, newSequentialCounter(ctrl), mockStorage, nil, nil) msg := entityqueue.NewMessage("test-queue/123", requestIDPayload(t, "test-queue/123"), "test-queue", nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -253,7 +254,7 @@ func TestController_Process_PublishFailure(t *testing.T) { request := testRequest() msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -270,7 +271,7 @@ func TestController_Process_CounterFailure(t *testing.T) { request := testRequest() msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -329,7 +330,7 @@ func TestController_Process_WithDependencies(t *testing.T) { controller := newTestController(t, ctrl, newSequentialCounter(ctrl), mockStorage, nil, nil) msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -381,7 +382,7 @@ func TestController_Process_AnalyzerSelectsSubset(t *testing.T) { controller := newTestController(t, ctrl, newSequentialCounter(ctrl), mockStorage, analyzer, nil) msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -410,7 +411,7 @@ func TestController_Process_AnalyzerFailure(t *testing.T) { controller := newTestController(t, ctrl, newSequentialCounter(ctrl), mockStorage, analyzer, nil) msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -466,7 +467,7 @@ func TestController_Process_HaltedShortCircuit(t *testing.T) { controller := newTestController(t, ctrl, cnt, mockStorage, nil, fmt.Errorf("should not publish")) msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -528,7 +529,7 @@ func TestController_Process_CASLostToCancel(t *testing.T) { ) msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -565,7 +566,7 @@ func TestController_Process_CASUnexpectedErrorPropagates(t *testing.T) { controller := newTestController(t, ctrl, newSequentialCounter(ctrl), mockStorage, nil, nil) msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -610,7 +611,7 @@ func TestController_Process_RecoveryAfterPriorCAS(t *testing.T) { controller := newTestController(t, ctrl, newSequentialCounter(ctrl), mockStorage, nil, nil) msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() diff --git a/submitqueue/orchestrator/controller/build/BUILD.bazel b/submitqueue/orchestrator/controller/build/BUILD.bazel index d4386494..9301f3ce 100644 --- a/submitqueue/orchestrator/controller/build/BUILD.bazel +++ b/submitqueue/orchestrator/controller/build/BUILD.bazel @@ -25,6 +25,7 @@ go_test( deps = [ "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//submitqueue/core/changeset/fake:go_default_library", diff --git a/submitqueue/orchestrator/controller/build/build_test.go b/submitqueue/orchestrator/controller/build/build_test.go index 7347ae29..3af0e2ae 100644 --- a/submitqueue/orchestrator/controller/build/build_test.go +++ b/submitqueue/orchestrator/controller/build/build_test.go @@ -24,6 +24,7 @@ import ( "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/platform/errs" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" changesetfake "github.com/uber/submitqueue/submitqueue/core/changeset/fake" @@ -129,7 +130,7 @@ func TestController_Process_Success(t *testing.T) { controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), nil) msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -201,7 +202,7 @@ func TestController_Process_TriggersWithBaseAndHead(t *testing.T) { controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, store, staticBuildRunnerFactory{r: br}, registry, topickey.TopicKeyBuild, "orchestrator-build") msg := entityqueue.NewMessage(headBatch.ID, batchIDPayload(t, headBatch.ID), headBatch.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -258,7 +259,7 @@ func TestController_Process_BuildStoreAlreadyExistsIsSwallowed(t *testing.T) { controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, store, staticBuildRunnerFactory{r: br}, registry, topickey.TopicKeyBuild, "orchestrator-build") msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -290,7 +291,7 @@ func TestController_Process_TriggerFailure(t *testing.T) { controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, store, staticBuildRunnerFactory{r: br}, registry, topickey.TopicKeyBuild, "orchestrator-build") msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -310,7 +311,7 @@ func TestController_Process_StorageFailure(t *testing.T) { controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), nil) msg := entityqueue.NewMessage("test-queue/batch/1", batchIDPayload(t, "test-queue/batch/1"), "test-queue", nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -327,7 +328,7 @@ func TestController_Process_PublishFailure(t *testing.T) { controller := newTestController(t, ctrl, store, buildfake.New(changesetfake.New()), fmt.Errorf("publish failed")) msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -374,7 +375,7 @@ func TestController_Process_HaltedShortCircuit(t *testing.T) { controller := newTestController(t, ctrl, store, br, fmt.Errorf("should not publish")) msg := entityqueue.NewMessage(batch.ID, batchIDPayload(t, batch.ID), batch.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() diff --git a/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel b/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel index 96799c87..58b23a5b 100644 --- a/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel +++ b/submitqueue/orchestrator/controller/buildsignal/BUILD.bazel @@ -25,6 +25,7 @@ go_test( deps = [ "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//submitqueue/core/topickey:go_default_library", diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go index affd0cbf..b1a144e0 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go +++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go @@ -24,6 +24,7 @@ import ( "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/platform/errs" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/submitqueue/core/topickey" @@ -98,7 +99,7 @@ func buildDelivery(t *testing.T, ctrl *gomock.Controller, b entity.Build) consum payload, err := entity.BuildID{ID: b.ID}.ToBytes() require.NoError(t, err) msg := entityqueue.NewMessage(b.ID, payload, b.BatchID, nil) - d := queuemock.NewMockDelivery(ctrl) + d := consumermock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(msg).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() return d @@ -276,7 +277,7 @@ func TestController_Process_MalformedPayload(t *testing.T) { h := newTestHarness(t, ctrl) msg := entityqueue.NewMessage("bad", []byte(`{"invalid"`), "batch-bad", nil) - d := queuemock.NewMockDelivery(ctrl) + d := consumermock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(msg).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() diff --git a/submitqueue/orchestrator/controller/cancel/BUILD.bazel b/submitqueue/orchestrator/controller/cancel/BUILD.bazel index 66f99b3f..c3e5f24d 100644 --- a/submitqueue/orchestrator/controller/cancel/BUILD.bazel +++ b/submitqueue/orchestrator/controller/cancel/BUILD.bazel @@ -25,6 +25,7 @@ go_test( deps = [ "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", diff --git a/submitqueue/orchestrator/controller/cancel/cancel_test.go b/submitqueue/orchestrator/controller/cancel/cancel_test.go index 79724a99..2c611b48 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel_test.go +++ b/submitqueue/orchestrator/controller/cancel/cancel_test.go @@ -24,6 +24,7 @@ import ( "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" @@ -65,7 +66,7 @@ func newController(t *testing.T, store storage.Storage, registry consumer.TopicR func newDelivery(t *testing.T, ctrl *gomock.Controller, payload []byte, partitionKey string) consumer.Delivery { msg := entityqueue.NewMessage("cancel-msg", payload, partitionKey, nil) - d := queuemock.NewMockDelivery(ctrl) + d := consumermock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(msg).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() return d diff --git a/submitqueue/orchestrator/controller/conclude/BUILD.bazel b/submitqueue/orchestrator/controller/conclude/BUILD.bazel index 13fa7175..4cff6b19 100644 --- a/submitqueue/orchestrator/controller/conclude/BUILD.bazel +++ b/submitqueue/orchestrator/controller/conclude/BUILD.bazel @@ -23,6 +23,7 @@ go_test( deps = [ "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//submitqueue/core/topickey:go_default_library", diff --git a/submitqueue/orchestrator/controller/conclude/conclude_test.go b/submitqueue/orchestrator/controller/conclude/conclude_test.go index d2d8c0e7..6fbbf1af 100644 --- a/submitqueue/orchestrator/controller/conclude/conclude_test.go +++ b/submitqueue/orchestrator/controller/conclude/conclude_test.go @@ -24,6 +24,7 @@ import ( "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/platform/errs" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/submitqueue/core/topickey" @@ -417,7 +418,7 @@ func TestController_Process(t *testing.T) { controller, _ := newTestController(t, ctrl, mockStorage, tt.expectLogPublish) msg := entityqueue.NewMessage(tt.batch.ID, batchIDPayload(t, tt.batch.ID), tt.batch.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -445,7 +446,7 @@ func TestController_Process_StorageFailure(t *testing.T) { controller, _ := newTestController(t, ctrl, mockStorage, false) msg := entityqueue.NewMessage("test-queue/batch/1", batchIDPayload(t, "test-queue/batch/1"), "test-queue", nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() diff --git a/submitqueue/orchestrator/controller/dlq/BUILD.bazel b/submitqueue/orchestrator/controller/dlq/BUILD.bazel index 08a52c22..0dda0190 100644 --- a/submitqueue/orchestrator/controller/dlq/BUILD.bazel +++ b/submitqueue/orchestrator/controller/dlq/BUILD.bazel @@ -43,6 +43,7 @@ go_test( "//api/runway/messagequeue/protopb:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//submitqueue/core/topickey:go_default_library", diff --git a/submitqueue/orchestrator/controller/dlq/request_test.go b/submitqueue/orchestrator/controller/dlq/request_test.go index 17fc0eab..e72b415e 100644 --- a/submitqueue/orchestrator/controller/dlq/request_test.go +++ b/submitqueue/orchestrator/controller/dlq/request_test.go @@ -22,7 +22,7 @@ import ( "github.com/stretchr/testify/require" queue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" - queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" @@ -168,9 +168,9 @@ func TestDLQRequestController_Process_EmptyIDFails(t *testing.T) { // newMockDelivery returns a MockDelivery wired up enough to be passed through // the DLQ controller Process flow. -func newMockDelivery(ctrl *gomock.Controller, payload []byte) *queuemock.MockDelivery { +func newMockDelivery(ctrl *gomock.Controller, payload []byte) *consumermock.MockDelivery { msg := queue.NewMessage("dlq-msg-1", payload, "", nil) - d := queuemock.NewMockDelivery(ctrl) + d := consumermock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(msg).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() d.EXPECT().Metadata().Return(map[string]string{ diff --git a/submitqueue/orchestrator/controller/merge/BUILD.bazel b/submitqueue/orchestrator/controller/merge/BUILD.bazel index 2e8710ab..d62738ee 100644 --- a/submitqueue/orchestrator/controller/merge/BUILD.bazel +++ b/submitqueue/orchestrator/controller/merge/BUILD.bazel @@ -31,6 +31,7 @@ go_test( "//platform/base/mergestrategy:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//submitqueue/core/topickey:go_default_library", diff --git a/submitqueue/orchestrator/controller/merge/merge_test.go b/submitqueue/orchestrator/controller/merge/merge_test.go index 39f6664e..81212c31 100644 --- a/submitqueue/orchestrator/controller/merge/merge_test.go +++ b/submitqueue/orchestrator/controller/merge/merge_test.go @@ -31,6 +31,7 @@ import ( "github.com/uber/submitqueue/platform/base/mergestrategy" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/platform/errs" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/submitqueue/core/topickey" @@ -45,9 +46,9 @@ func batchIDPayload(t *testing.T, id string) []byte { return payload } -func newDelivery(t *testing.T, ctrl *gomock.Controller, batchID, partitionKey string) *queuemock.MockDelivery { +func newDelivery(t *testing.T, ctrl *gomock.Controller, batchID, partitionKey string) *consumermock.MockDelivery { msg := entityqueue.NewMessage(batchID, batchIDPayload(t, batchID), partitionKey, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() return delivery diff --git a/submitqueue/orchestrator/controller/mergeconflictsignal/BUILD.bazel b/submitqueue/orchestrator/controller/mergeconflictsignal/BUILD.bazel index f349b363..0d3b668a 100644 --- a/submitqueue/orchestrator/controller/mergeconflictsignal/BUILD.bazel +++ b/submitqueue/orchestrator/controller/mergeconflictsignal/BUILD.bazel @@ -29,6 +29,7 @@ go_test( "//api/runway/messagequeue/protopb:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", diff --git a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal_test.go b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal_test.go index 56f289d3..c847b369 100644 --- a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal_test.go +++ b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal_test.go @@ -25,6 +25,7 @@ import ( runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" @@ -39,8 +40,8 @@ func resultPayload(t *testing.T, res runwaymq.MergeResult) []byte { return payload } -func newDelivery(ctrl *gomock.Controller, msg entityqueue.Message) *queuemock.MockDelivery { - d := queuemock.NewMockDelivery(ctrl) +func newDelivery(ctrl *gomock.Controller, msg entityqueue.Message) *consumermock.MockDelivery { + d := consumermock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(msg).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() return d diff --git a/submitqueue/orchestrator/controller/mergesignal/BUILD.bazel b/submitqueue/orchestrator/controller/mergesignal/BUILD.bazel index 0a0e081b..78018f23 100644 --- a/submitqueue/orchestrator/controller/mergesignal/BUILD.bazel +++ b/submitqueue/orchestrator/controller/mergesignal/BUILD.bazel @@ -28,6 +28,7 @@ go_test( "//api/runway/messagequeue/protopb:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", diff --git a/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go b/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go index 7ca96911..a872856c 100644 --- a/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go +++ b/submitqueue/orchestrator/controller/mergesignal/mergesignal_test.go @@ -25,6 +25,7 @@ import ( runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" @@ -44,8 +45,8 @@ func resultPayload(t *testing.T, res runwaymq.MergeResult) []byte { return payload } -func newDelivery(ctrl *gomock.Controller, msg entityqueue.Message) *queuemock.MockDelivery { - d := queuemock.NewMockDelivery(ctrl) +func newDelivery(ctrl *gomock.Controller, msg entityqueue.Message) *consumermock.MockDelivery { + d := consumermock.NewMockDelivery(ctrl) d.EXPECT().Message().Return(msg).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() return d diff --git a/submitqueue/orchestrator/controller/speculate/BUILD.bazel b/submitqueue/orchestrator/controller/speculate/BUILD.bazel index 39e36977..830c8b73 100644 --- a/submitqueue/orchestrator/controller/speculate/BUILD.bazel +++ b/submitqueue/orchestrator/controller/speculate/BUILD.bazel @@ -24,6 +24,7 @@ go_test( deps = [ "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//submitqueue/core/topickey:go_default_library", diff --git a/submitqueue/orchestrator/controller/speculate/speculate_test.go b/submitqueue/orchestrator/controller/speculate/speculate_test.go index 85168da8..8bdd3e23 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate_test.go +++ b/submitqueue/orchestrator/controller/speculate/speculate_test.go @@ -24,6 +24,7 @@ import ( "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/platform/errs" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/submitqueue/core/topickey" @@ -84,7 +85,7 @@ func newTestController(t *testing.T, ctrl *gomock.Controller, store *storagemock // runProcess builds a delivery for batchID and invokes Process once. func runProcess(t *testing.T, ctrl *gomock.Controller, controller *Controller, batchID string) error { msg := entityqueue.NewMessage(batchID, batchIDPayload(t, batchID), "test-queue", nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() return controller.Process(context.Background(), delivery) @@ -612,7 +613,7 @@ func TestController_Process_BadPayload(t *testing.T) { controller := newTestController(t, ctrl, store, nil) msg := entityqueue.NewMessage("anything", []byte("not-json"), "test-queue", nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() diff --git a/submitqueue/orchestrator/controller/start/BUILD.bazel b/submitqueue/orchestrator/controller/start/BUILD.bazel index 3f0aaa19..960d163d 100644 --- a/submitqueue/orchestrator/controller/start/BUILD.bazel +++ b/submitqueue/orchestrator/controller/start/BUILD.bazel @@ -27,6 +27,7 @@ go_test( "//platform/base/mergestrategy:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//submitqueue/core/topickey:go_default_library", diff --git a/submitqueue/orchestrator/controller/start/start_test.go b/submitqueue/orchestrator/controller/start/start_test.go index 733ae9f6..cea4d17a 100644 --- a/submitqueue/orchestrator/controller/start/start_test.go +++ b/submitqueue/orchestrator/controller/start/start_test.go @@ -26,6 +26,7 @@ import ( "github.com/uber/submitqueue/platform/base/mergestrategy" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/platform/errs" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/submitqueue/core/topickey" @@ -78,12 +79,12 @@ func newMockStorage(ctrl *gomock.Controller) *storagemock.MockStorage { } // makeDelivery builds a MockDelivery wrapping a serialized LandRequest. -func makeDelivery(t *testing.T, ctrl *gomock.Controller, lr entity.LandRequest) *queuemock.MockDelivery { +func makeDelivery(t *testing.T, ctrl *gomock.Controller, lr entity.LandRequest) *consumermock.MockDelivery { payload, err := lr.ToBytes() require.NoError(t, err) msg := entityqueue.NewMessage(lr.ID, payload, lr.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() return delivery @@ -119,7 +120,7 @@ func TestController_Process_InvalidJSON(t *testing.T) { invalidPayload := []byte(`{"invalid": json"}`) msg := entityqueue.NewMessage("invalid-msg", invalidPayload, "partition1", nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() diff --git a/submitqueue/orchestrator/controller/validate/BUILD.bazel b/submitqueue/orchestrator/controller/validate/BUILD.bazel index 3a0eb238..8f2a923f 100644 --- a/submitqueue/orchestrator/controller/validate/BUILD.bazel +++ b/submitqueue/orchestrator/controller/validate/BUILD.bazel @@ -34,6 +34,7 @@ go_test( "//platform/base/mergestrategy:go_default_library", "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/consumer/mock:go_default_library", "//platform/errs:go_default_library", "//platform/extension/messagequeue/mock:go_default_library", "//submitqueue/core/topickey:go_default_library", diff --git a/submitqueue/orchestrator/controller/validate/validate_test.go b/submitqueue/orchestrator/controller/validate/validate_test.go index 615638a0..30905071 100644 --- a/submitqueue/orchestrator/controller/validate/validate_test.go +++ b/submitqueue/orchestrator/controller/validate/validate_test.go @@ -28,6 +28,7 @@ import ( "github.com/uber/submitqueue/platform/base/mergestrategy" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/platform/errs" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/submitqueue/core/topickey" @@ -161,7 +162,7 @@ func TestController_Process_Success(t *testing.T) { controller := newTestController(t, ctrl, store, newMockChangeStore(ctrl), nil) msg := entityqueue.NewMessage("test-queue/123", requestIDPayload(t, request.ID), "test-queue", nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -209,7 +210,7 @@ func TestController_Process_PublishesCheckToRunway(t *testing.T) { controller := NewController(logger, tally.NoopScope, store, registry, cpFactory, nil, runwaymq.TopicKeyMergeConflictCheck, topickey.TopicKeyValidate, "orchestrator-validate") msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -268,7 +269,7 @@ func TestController_Process_ClaimsChangeRecordsWithDetails(t *testing.T) { controller := newTestController(t, ctrl, store, cs, nil) msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -286,7 +287,7 @@ func TestController_Process_StorageFailure(t *testing.T) { controller := newTestController(t, ctrl, store, newMockChangeStore(ctrl), nil) msg := entityqueue.NewMessage("test-queue/123", requestIDPayload(t, "test-queue/123"), "test-queue", nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -311,7 +312,7 @@ func TestController_Process_PublishFailure(t *testing.T) { controller := newTestController(t, ctrl, store, newMockChangeStore(ctrl), fmt.Errorf("publish failed")) msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -489,7 +490,7 @@ func TestController_Process_DuplicateDetection(t *testing.T) { controller := newTestController(t, ctrl, store, cs, nil) msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -529,7 +530,7 @@ func TestController_Process_ChangeStoreQueryFailure(t *testing.T) { controller := newTestController(t, ctrl, store, cs, nil) msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -568,7 +569,7 @@ func TestController_Process_TerminalShortCircuit(t *testing.T) { controller := newTestController(t, ctrl, store, cs, fmt.Errorf("should not publish")) msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -616,7 +617,7 @@ func TestController_Process_CustomValidatorPasses(t *testing.T) { controller := NewController(logger, tally.NoopScope, store, registry, cpFactory, mockValidatorFactory, runwaymq.TopicKeyMergeConflictCheck, topickey.TopicKeyValidate, "orchestrator-validate") msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -675,7 +676,7 @@ func TestController_Process_CustomValidatorFails(t *testing.T) { controller := NewController(logger, tally.NoopScope, store, registry, cpFactory, mockValidatorFactory, runwaymq.TopicKeyMergeConflictCheck, topickey.TopicKeyValidate, "orchestrator-validate") msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() @@ -726,7 +727,7 @@ func TestController_Process_CustomValidatorFailure_TerminationPublishFails(t *te controller := NewController(zaptest.NewLogger(t).Sugar(), tally.NoopScope, store, registry, cpFactory, mockValidatorFactory, runwaymq.TopicKeyMergeConflictCheck, topickey.TopicKeyValidate, "orchestrator-validate") msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil) - delivery := queuemock.NewMockDelivery(ctrl) + delivery := consumermock.NewMockDelivery(ctrl) delivery.EXPECT().Message().Return(msg).AnyTimes() delivery.EXPECT().Attempt().Return(1).AnyTimes() diff --git a/test/integration/extension/messagequeue/mysql/queue_test.go b/test/integration/extension/messagequeue/mysql/queue_test.go index ed006a6e..e4c35b10 100644 --- a/test/integration/extension/messagequeue/mysql/queue_test.go +++ b/test/integration/extension/messagequeue/mysql/queue_test.go @@ -2055,6 +2055,135 @@ func (s *SQLQueueIntegrationSuite) TestNackDoesNotBlockOtherMessages() { t.Logf("Verified: nacked message did not block subsequent messages") } +// TestPostponeBlocksPartitionUntilDue verifies the postpone barrier: while a +// postponed message waits out its delay, later offsets in the partition are +// not delivered; when due, the postponed message redelivers first, in order, +// and as a first attempt (postponing does not consume retry budget). +func (s *SQLQueueIntegrationSuite) TestPostponeBlocksPartitionUntilDue() { + t := s.T() + + signalCh := make(chan queueMySQL.HookSignal, 100) + q, err := queueMySQL.NewQueue(queueMySQL.Params{ + DB: s.db, + Logger: zaptest.NewLogger(t), + MetricsScope: tally.NoopScope, + OnSignal: signalCh, + }) + require.NoError(t, err) + defer q.Close() + + topic := "postpone_barrier_topic" + partition := "postpone-part" + + // Subscribe with batch=10 so the barrier — not the batch size — is what + // keeps later offsets back. + subConfig := extqueue.DefaultSubscriptionConfig("worker-1", "postpone-cg") + subConfig.PollIntervalMs = 50 + subConfig.BatchSize = 10 + deliveryCh, err := q.Subscriber().Subscribe(s.ctx, topic, subConfig) + require.NoError(t, err) + + // Publish the first message alone, receive it, and postpone it. The + // barrier must be in place before the later messages exist — deliveries + // already fetched into the in-memory buffer are past the barrier by + // design (it acts at the fetch layer). + 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) + postponeDelay := 2 * time.Second + require.NoError(t, d1.Postpone(s.ctx, postponeDelay.Milliseconds())) + t.Logf("Postponed msg-1 for %s", postponeDelay) + + // Publish two more messages behind the postponed one + 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)) + } + + // Barrier: messages 2 and 3 must not be delivered while msg-1 waits — + // the opposite of the nacked case above. + assertNoDelivery(t, deliveryCh, signalCh, queueMySQL.SignalDeliveryCheck, 3) + t.Logf("Confirmed: partition blocked behind postponed msg-1") + + // When due, msg-1 redelivers first, in order, as a fresh first attempt + d1again := receive(t, deliveryCh) + assert.Equal(t, "msg-1", d1again.Message().ID) + assert.Equal(t, 1, d1again.Attempt(), "postponed redelivery must not consume retry budget") + require.NoError(t, d1again.Ack(s.ctx)) + t.Logf("Received postponed msg-1 again as attempt 1") + + d2 := receive(t, deliveryCh) + assert.Equal(t, "msg-2", d2.Message().ID) + require.NoError(t, d2.Ack(s.ctx)) + + d3 := receive(t, deliveryCh) + assert.Equal(t, "msg-3", d3.Message().ID) + require.NoError(t, d3.Ack(s.ctx)) + + t.Logf("Verified: postponed message acted as a barrier and redelivered in order") +} + +// TestPostponeResetsRetryBudget verifies a postpone does not weaken the DLQ +// backstop: after a postpone the message restarts as attempt 1, and real +// failures still dead-letter at MaxAttempts. +func (s *SQLQueueIntegrationSuite) TestPostponeResetsRetryBudget() { + t := s.T() + + signalCh := make(chan queueMySQL.HookSignal, 100) + q, err := queueMySQL.NewQueue(queueMySQL.Params{ + DB: s.db, + Logger: zaptest.NewLogger(t), + MetricsScope: tally.NoopScope, + OnSignal: signalCh, + }) + require.NoError(t, err) + defer q.Close() + + topic := "postpone_budget_topic" + + subConfig := testSubConfig("worker-1", "postpone-budget-cg") + subConfig.Retry.MaxAttempts = 2 + subConfig.DLQ.Enabled = true + + deliveryChan, err := q.Subscriber().Subscribe(s.ctx, topic, subConfig) + require.NoError(t, err) + + msg := entityqueue.NewMessage("wait-then-poison", []byte("payload"), "partition-1", nil) + require.NoError(t, q.Publisher().Publish(s.ctx, topic, msg)) + + // First delivery: postpone briefly — a deliberate wait, not a failure + d := receive(t, deliveryChan) + assert.Equal(t, 1, d.Attempt()) + require.NoError(t, d.Postpone(s.ctx, 100)) + t.Logf("Postponed message on first delivery") + + // The post-postpone redelivery restarts at attempt 1; now fail for real + // MaxAttempts times. + for attempt := 1; attempt <= subConfig.Retry.MaxAttempts; attempt++ { + 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)) + t.Logf("Attempt %d: nacked", delivery.Attempt()) + } + + // The message dead-letters despite the earlier postpone + assertNoDelivery(t, deliveryChan, signalCh, queueMySQL.SignalDeliveryCheck, 3) + + dlqTopic := topic + subConfig.DLQ.TopicSuffix + dlqConfig := extqueue.DefaultSubscriptionConfig("worker-1", "postpone-budget-cg") + dlqDeliveryChan, err := q.Subscriber().Subscribe(s.ctx, dlqTopic, dlqConfig) + require.NoError(t, err) + + dlqDelivery := receive(t, dlqDeliveryChan) + assert.Equal(t, "wait-then-poison", dlqDelivery.Message().ID) + require.NoError(t, dlqDelivery.Ack(s.ctx)) + + t.Logf("Verified: postpone reset the budget but real failures still dead-lettered") +} + // TestBatchSizeOneStrictSerialization verifies that with batchSize=1, messages // within a partition are processed strictly in order — only one message is // in-flight at a time.