README
¶
DLQ Reconciliation Controllers
This package contains the controllers that drain each primary pipeline topic's {topic}_dlq companion and reconcile the affected request or batch into a terminal failed state. They are wired alongside the primary controllers in service/submitqueue/orchestrator/server/main.go.
Design principles
The DLQ is the final destination. A message in {topic}_dlq has already failed the primary controller's retry budget (Retry.MaxAttempts for a retryable error) or surfaced as non-retryable on the first attempt. There is no further dead-letter beyond this point — every DLQ subscription is configured with DLQ.Enabled = false, so a DLQ controller's outcome is either "reconciled" or "the message keeps coming back". A genuinely unprocessable DLQ message (e.g. a malformed payload that no controller can ever decode) must be removed by an operator from the queue manually.
The implementation is deliberately the simplest and most reliable thing that can work. Each controller does the same three steps: decode the payload to recover the affected RequestID or BatchID, fetch the entity, transition it to its terminal failed state with the same optimistic-locking CAS the primary pipeline uses. No business logic, no branching on the original error, no per-stage cleanup. Adding behaviour here trades off the one property the DLQ has to provide — convergence.
Reconcile only; do not attempt to recover. A DLQ controller never tries to re-run the failed stage, retry the original action against an external dependency, or repair a partially-completed transition. The original controller already failed; the DLQ controller's only job is to make sure the request and batch state stop saying "in progress" and start saying "failed", so the gateway reports an honest answer and downstream tooling can move on. Recovery, when it is appropriate, is a separate concern handled by an operator or a future reconciliation job — not by this code path.
Convergence guarantee
DLQ consumers are wired with errs.AlwaysRetryableProcessor and a very high Retry.MaxAttempts (currently 1000). Together with DLQ.Enabled = false on the DLQ subscription itself, this means any non-nil error returned from a DLQ controller — including a plain unclassified infra error — is forced retryable and redelivered rather than silently dropped. The combination is "always retryable + bounded-but-effectively-infinite attempts" and is the property the package relies on for convergence.
The recognised error condition is handled explicitly in dlq.go:
storage.ErrNotFound→ logged at warn and treated as success. The request or batch never persisted; there is nothing to reconcile.
Everything else — including storage.ErrVersionMismatch on the CAS — is returned without controller-level classification and redelivered until it either succeeds or hits the attempt cap. ErrVersionMismatch is already retryable at its declaration, while the always-retryable processor makes every other non-nil reconciliation error retryable too.
Request log entries are published to Gateway
When a DLQ controller transitions a request to RequestStateError, it publishes the terminal request log to the log topic. Gateway consumes that topic and remains the sole writer of the request log and public projections. A publish failure leaves the DLQ delivery unacknowledged so redelivery retries the same idempotent log event.
Controller mapping
Two controller shapes cover the eleven primary pipeline topics:
| Controller | Topics | Decoded ID | Terminal state |
|---|---|---|---|
NewDLQRequestController |
start, validate, batch, cancel, log |
RequestID |
RequestStateError |
NewDLQBatchController |
speculate, build, merge, conclude |
BatchID |
BatchStateFailed + fan-out to member requests as RequestStateError |
buildsignal carries a Build payload and has its own small dedicated controller. The split exists because the DLQ message payload shape mirrors the primary topic's payload shape (the queue framework preserves bytes verbatim under the _dlq topic name), so the decoder is what changes per topic — not the reconciliation step. The package-level RequestIDDecoder interface plus DecodeLandRequestID / DecodeCancelRequestID / DecodeRequestID cover the three payload shapes used by request-scoped topics.
Idempotency and concurrent activity
Reconciliation is safe to run more than once for the same message:
- A request already in
RequestStateErrorskips the state-transition CAS but republishes its terminal log so redelivery repairs an earlier publish failure. Requests in a different terminal state are left unchanged. - A batch already in
BatchStateFailedstill fans out to member requests, because a previous attempt may have transitioned the batch but crashed before completing the fan-out. Batches in a different terminal state are left unchanged. - Per-request fan-out is itself idempotent via
failRequest. - A request in
RequestStateCancellingis reconciled toRequestStateError, not left in place: DLQ means the pipeline failed to converge, so we cannot confirm the cancel completed cleanly. Writing Error is the honest signal.
See also
core/errs/README.md— the error-processing framework, includingAlwaysRetryableProcessorand the choice of processor for primary vs. DLQ consumers.core/consumer/README.md— how the consumer applies the processor to controller errors and decides ack/nack/reject.doc/rfc/submitqueue/workflow.md— the per-stage primary pipeline that the DLQ companions mirror.
Documentation
¶
Overview ¶
Package dlq contains controllers that consume messages from per-topic dead-letter queues and reconcile the affected request and batch entities into a terminal failed state.
Background. The consumer framework moves a message to its DLQ after the controller for the original topic returns a non-retryable error or exhausts retries on a retryable error. Without DLQ reconciliation the affected request would remain stuck in a non-terminal state (e.g. Validated, Batched, Processing) forever — the gateway would still report it as "in progress" even though no pipeline stage is going to advance it.
Reconciliation strategy. Each DLQ topic carries the same payload as its originating topic (the queue framework preserves the bytes verbatim under a new `{topic}_dlq` name). The DLQ controllers decode that payload to recover the affected request or batch, then transition it to a terminal failed state — Error for requests, Failed for batches — with an idempotent optimistic-locking write so concurrent activity (a late merge, a cancel race) wins cleanly. Batch failures also fan out to the member requests so the gateway no longer reports them as in-progress.
Index ¶
- func DecodeCancelRequestID(payload []byte) (entity.RequestID, error)
- func DecodeLandRequestID(payload []byte) (entity.RequestID, error)
- func DecodeRequestID(payload []byte) (entity.RequestID, error)
- func NewDLQBatchController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, ...) consumer.Controller
- func NewDLQBuildSignalController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, ...) consumer.Controller
- func NewDLQLogController(logger *zap.SugaredLogger, scope tally.Scope, topicKey consumer.TopicKey, ...) consumer.Controller
- func NewDLQMergeConflictSignalController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, ...) consumer.Controller
- func NewDLQMergeSignalController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, ...) consumer.Controller
- func NewDLQRequestController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, ...) consumer.Controller
- func NewDLQSpeculateController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, ...) consumer.Controller
- func TopicKey(main consumer.TopicKey) consumer.TopicKey
- type RequestIDDecoder
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func DecodeCancelRequestID ¶
DecodeCancelRequestID extracts the request ID from a CancelRequest payload (the shape used by the cancel topic).
func DecodeLandRequestID ¶
DecodeLandRequestID extracts the request ID from a LandRequest payload (the shape used by the start topic).
func DecodeRequestID ¶
DecodeRequestID extracts the request ID from a RequestID payload (the shape used by the validate and batch topics).
func NewDLQBatchController ¶
func NewDLQBatchController( logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, ) consumer.Controller
NewDLQBatchController builds a DLQ controller for a batch-scoped topic. topicKey must be the DLQ topic key (typically TopicKey(primary)).
func NewDLQBuildSignalController ¶
func NewDLQBuildSignalController( logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, ) consumer.Controller
NewDLQBuildSignalController builds a DLQ controller for the buildsignal topic.
func NewDLQLogController ¶
func NewDLQLogController( logger *zap.SugaredLogger, scope tally.Scope, topicKey consumer.TopicKey, consumerGroup string, ) consumer.Controller
NewDLQLogController builds a DLQ controller for the log topic.
func NewDLQMergeConflictSignalController ¶
func NewDLQMergeConflictSignalController( logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, ) consumer.Controller
NewDLQMergeConflictSignalController builds a DLQ controller for the mergeconflictsignal topic.
func NewDLQMergeSignalController ¶
func NewDLQMergeSignalController( logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, ) consumer.Controller
NewDLQMergeSignalController builds a DLQ controller for the mergesignal topic.
func NewDLQRequestController ¶
func NewDLQRequestController( logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, registry consumer.TopicRegistry, decode RequestIDDecoder, topicKey consumer.TopicKey, consumerGroup string, ) consumer.Controller
NewDLQRequestController builds a DLQ controller for a request-scoped topic. topicKey must be the DLQ topic key (typically TopicKey(primary)); decode must match the payload shape of the primary topic this DLQ drains.
func NewDLQSpeculateController ¶
func NewDLQSpeculateController( logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, ) consumer.Controller
NewDLQSpeculateController builds the DLQ controller for the speculate topic. topicKey must be the DLQ topic key (typically TopicKey(primary)).
func TopicKey ¶
TopicKey returns the DLQ topic key for the given primary pipeline topic. The returned key is meant to be used both when registering the DLQ topic with the topic registry and when the corresponding DLQ controller advertises its TopicKey(). It is exported so the orchestrator wiring layer can build matching pairs without duplicating the suffix literal.
Types ¶
type RequestIDDecoder ¶
RequestIDDecoder extracts the affected request's identity — its ID and its queue — from the raw payload bytes of a DLQ message. Different primary topics carry different payload shapes (LandRequest on start, CancelRequest on cancel, RequestID on validate / batch), so the caller injects the right decoder for the topic being reconciled. Returning an empty ID is treated as a decode failure.
Source Files