|
| 1 | +// Copyright (c) 2025 Uber Technologies, Inc. |
| 2 | +// |
| 3 | +// Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | +// you may not use this file except in compliance with the License. |
| 5 | +// You may obtain a copy of the License at |
| 6 | +// |
| 7 | +// http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | +// |
| 9 | +// Unless required by applicable law or agreed to in writing, software |
| 10 | +// distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | +// See the License for the specific language governing permissions and |
| 13 | +// limitations under the License. |
| 14 | + |
| 15 | +// Package dlq contains controllers that consume messages from a pipeline stage's |
| 16 | +// dead-letter topic and reconcile the affected request into a terminal state. |
| 17 | +// |
| 18 | +// Background. The consumer framework moves a message to its DLQ after the controller |
| 19 | +// for the original topic returns a non-retryable error or exhausts retries on a |
| 20 | +// retryable error. Without DLQ reconciliation the affected request would remain stuck |
| 21 | +// in a non-terminal state (accepted, processing) forever — a caller gating deployments |
| 22 | +// on greenness would see that indistinguishably from "not yet validated" (see |
| 23 | +// doc/rfc/stovepipe/workflow.md#fail-closed-on-unprocessable-work). |
| 24 | +// |
| 25 | +// Reconciliation strategy. Each DLQ topic carries the same payload as its originating |
| 26 | +// topic (the queue framework preserves the bytes verbatim under a new `{topic}_dlq` |
| 27 | +// name). The DLQ controller decodes that payload to recover the affected request, then |
| 28 | +// transitions it to RequestStateRecordedNotGreen — the conservative not-green verdict |
| 29 | +// for gating (see entity.RequestState) — with an idempotent optimistic-locking write so |
| 30 | +// concurrent activity (a late successful pipeline transition) wins cleanly. If the request had |
| 31 | +// already been admitted (processing) and was holding a concurrency slot, the |
| 32 | +// reconciler also releases it by CAS-decrementing the queue's in_flight_count, per |
| 33 | +// doc/rfc/stovepipe/steps/process.md#in_flight_count-integrity. |
| 34 | +package dlq |
| 35 | + |
| 36 | +import ( |
| 37 | + "context" |
| 38 | + "errors" |
| 39 | + "fmt" |
| 40 | + |
| 41 | + "github.com/uber/submitqueue/platform/consumer" |
| 42 | + "github.com/uber/submitqueue/stovepipe/entity" |
| 43 | + "github.com/uber/submitqueue/stovepipe/extension/storage" |
| 44 | + "go.uber.org/zap" |
| 45 | +) |
| 46 | + |
| 47 | +// topicSuffix is appended to a primary topic key to derive the corresponding DLQ topic |
| 48 | +// key. The queue extension's DefaultSubscriptionConfig also uses "_dlq" as the DLQ |
| 49 | +// topic suffix; keeping both in sync is intentional so a registered DLQ subscription's |
| 50 | +// topic name matches the controller's TopicKey(). |
| 51 | +const topicSuffix = "_dlq" |
| 52 | + |
| 53 | +// TopicKey returns the DLQ topic key for the given primary pipeline topic. It is |
| 54 | +// exported so the stovepipe wiring layer can build matching pairs without duplicating |
| 55 | +// the suffix literal. |
| 56 | +func TopicKey(main consumer.TopicKey) consumer.TopicKey { |
| 57 | + return consumer.TopicKey(string(main) + topicSuffix) |
| 58 | +} |
| 59 | + |
| 60 | +// failRequest transitions request to RequestStateRecordedNotGreen if it is not already |
| 61 | +// in a terminal state. If the request had reached RequestStateProcessing — meaning process's |
| 62 | +// admit step already CAS-incremented the queue's in_flight_count for it — the queue's |
| 63 | +// slot is released first. Queue and Request are separate entities with no cross-entity |
| 64 | +// transaction, so the two writes cannot be atomic and the ordering picks which crash |
| 65 | +// failure mode we accept: a crash between the writes leaves the request non-terminal, |
| 66 | +// redelivery re-runs reconciliation, and releaseSlot (which tracks no per-request slot |
| 67 | +// ownership) decrements again — transiently over-admitting by one slot until the |
| 68 | +// under-count re-converges at releaseSlot's zero clamp. The reverse order would leak |
| 69 | +// the slot instead: redelivery skips terminal requests, permanently shrinking the |
| 70 | +// queue's capacity toward a wedge. Over-admission is the failure mode we prefer. See |
| 71 | +// doc/rfc/stovepipe/steps/process.md#in_flight_count-integrity for the broader |
| 72 | +// counter-drift story. |
| 73 | +func failRequest(ctx context.Context, store storage.Storage, logger *zap.SugaredLogger, requestID string) error { |
| 74 | + request, err := store.GetRequestStore().Get(ctx, requestID) |
| 75 | + if err != nil { |
| 76 | + if errors.Is(err, storage.ErrNotFound) { |
| 77 | + logger.Warnw("dlq reconcile: request not found, skipping", |
| 78 | + "request_id", requestID, |
| 79 | + ) |
| 80 | + return nil |
| 81 | + } |
| 82 | + return fmt.Errorf("failed to get request %s: %w", requestID, err) |
| 83 | + } |
| 84 | + |
| 85 | + if request.State.IsTerminal() { |
| 86 | + logger.Infow("dlq reconcile: request already terminal, skipping", |
| 87 | + "request_id", requestID, |
| 88 | + "state", string(request.State), |
| 89 | + ) |
| 90 | + return nil |
| 91 | + } |
| 92 | + |
| 93 | + if request.State == entity.RequestStateProcessing { |
| 94 | + if err := releaseSlot(ctx, store, logger, request.Queue); err != nil { |
| 95 | + return fmt.Errorf("failed to release queue slot for request %s: %w", requestID, err) |
| 96 | + } |
| 97 | + } |
| 98 | + |
| 99 | + updated := request |
| 100 | + updated.State = entity.RequestStateRecordedNotGreen |
| 101 | + newVersion := request.Version + 1 |
| 102 | + if err := store.GetRequestStore().Update(ctx, updated, request.Version, newVersion); err != nil { |
| 103 | + return fmt.Errorf("failed to update request %s state to recorded_not_green: %w", requestID, err) |
| 104 | + } |
| 105 | + logger.Infow("dlq reconcile: request forced terminal not-green", |
| 106 | + "request_id", requestID, |
| 107 | + "previous_state", string(request.State), |
| 108 | + ) |
| 109 | + return nil |
| 110 | +} |
| 111 | + |
| 112 | +// releaseSlot CAS-decrements the queue's in_flight_count, retrying on version |
| 113 | +// conflicts, mirroring process.Controller's own CAS-retry loop for queue updates. |
| 114 | +func releaseSlot(ctx context.Context, store storage.Storage, logger *zap.SugaredLogger, queueName string) error { |
| 115 | + queueStore := store.GetQueueStore() |
| 116 | + |
| 117 | + for { |
| 118 | + queueRow, err := queueStore.Get(ctx, queueName) |
| 119 | + if err != nil { |
| 120 | + if errors.Is(err, storage.ErrNotFound) { |
| 121 | + logger.Warnw("dlq reconcile: queue not found, skipping slot release", |
| 122 | + "queue", queueName, |
| 123 | + ) |
| 124 | + return nil |
| 125 | + } |
| 126 | + return fmt.Errorf("failed to get queue %s: %w", queueName, err) |
| 127 | + } |
| 128 | + |
| 129 | + if queueRow.InFlightCount <= 0 { |
| 130 | + logger.Warnw("dlq reconcile: queue in_flight_count already at zero, skipping slot release", |
| 131 | + "queue", queueName, |
| 132 | + ) |
| 133 | + return nil |
| 134 | + } |
| 135 | + |
| 136 | + updated := queueRow |
| 137 | + updated.InFlightCount-- |
| 138 | + newVersion := queueRow.Version + 1 |
| 139 | + if err := queueStore.Update(ctx, updated, queueRow.Version, newVersion); err != nil { |
| 140 | + if errors.Is(err, storage.ErrVersionMismatch) { |
| 141 | + continue |
| 142 | + } |
| 143 | + return fmt.Errorf("failed to release slot for queue %s: %w", queueName, err) |
| 144 | + } |
| 145 | + logger.Infow("dlq reconcile: released queue slot", |
| 146 | + "queue", queueName, |
| 147 | + "in_flight_count", updated.InFlightCount, |
| 148 | + ) |
| 149 | + return nil |
| 150 | + } |
| 151 | +} |
0 commit comments