feat(stovepipe): add DLQ reconciliation for the process stage#347
Open
gsturges wants to merge 2 commits into
Open
feat(stovepipe): add DLQ reconciliation for the process stage#347gsturges wants to merge 2 commits into
gsturges wants to merge 2 commits into
Conversation
gsturges
requested review from
a team,
behinddwalls and
sbalabanov
as code owners
mnoah1
reviewed
Jul 14, 2026
A request whose process-stage message exhausts retries would previously sit in a non-terminal state forever, indistinguishable from "not yet validated" to callers gating on greenness. Per the fail-closed policy in doc/rfc/stovepipe/workflow.md, such requests must be driven to a conservative not-green terminal state instead. - Add RequestStateFailed terminal state and IsRequestStateTerminal helper to the entity package. - Add stovepipe/controller/dlq: a consumer.Controller that drains the process_dlq topic and CAS-transitions the affected request to failed, releasing the queue's in_flight_count slot first when the request had already been admitted (processing), per the in_flight_count integrity rule in doc/rfc/stovepipe/steps/process.md. - Wire a second consumer in the stovepipe server using errs.AlwaysRetryableProcessor, register the process_dlq topic with DLQ disabled (no cascade) and raised retry attempts, and start/stop the two consumers in the safe order (DLQ up first, drained last). Reconciliation is idempotent: already-terminal requests are skipped, and optimistic locking lets a late legitimate pipeline transition win. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Review feedback on the process DLQ PR: the DLQ subscription overrides (DLQ.Enabled=false, MaxAttempts=1000) were duplicated across wiring files, and two doc comments in the DLQ reconciler claimed behavior the code does not have (release-first ordering prevents double release; decode errors are non-retryable under AlwaysRetryableProcessor). - platform/extension/messagequeue/subscription_config.go: add DLQSubscriptionConfig helper owning the DLQ-consumer subscription policy - platform/extension/messagequeue/subscription_config_test.go: test the helper's contract - service/stovepipe/server/main.go: use the helper - service/submitqueue/orchestrator/server/main.go: use the helper - stovepipe/controller/dlq/dlq.go: correct failRequest comment to state the real crash trade-off (double release vs leaked slot) - stovepipe/controller/dlq/request.go: correct decode-error comment — retried deliberately to heal deployment skew, bounded by MaxAttempts Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
gsturges
force-pushed
the
gsturges/stovepipe-process-dlq
branch
from
587410a to
78d3d91
Compare
Contributor
Author
|
Rebased onto main. Besides the review feedback already discussed (the The reconciler now forces
Reconciliation behavior is otherwise unchanged (idempotent terminal-skip, slot release before the terminal CAS, always-retryable pairing). 🤖 Generated with Claude Code |
mnoah1
approved these changes
Jul 15, 2026
| func DLQSubscriptionConfig(subscriberName, consumerGroup string) SubscriptionConfig { | ||
| config := DefaultSubscriptionConfig(subscriberName, consumerGroup) | ||
| config.DLQ.Enabled = false | ||
| config.Retry.MaxAttempts = 1000 |
Collaborator
There was a problem hiding this comment.
should we just add -1? thoughts? i don't recall if messaging queue support that symantix but i think it might
behinddwalls
approved these changes
Jul 15, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
A request whose
process-stage message exhausts retries would previously sit in a non-terminal state forever — indistinguishable from "not yet validated" to callers gating deployments on greenness. Per the fail-closed policy in doc/rfc/stovepipe/workflow.md, such requests must be driven to a conservative not-green terminal state instead.RequestStateFailedterminal state andIsRequestStateTerminalhelper.stovepipe/controller/dlqpackage: aconsumer.Controllerthat drains theprocess_dlqtopic and CAS-transitions the affected request tofailed. If the request had already been admitted (processing), the queue'sin_flight_countslot is released first, per the integrity rule in doc/rfc/stovepipe/steps/process.md.errs.AlwaysRetryableProcessor(a reconciliation failure retries in place — the DLQ subscription is a final destination withDLQ.Enabled = false, so nothing can cascade to a_dlq_dlq);process_dlqregistered in the topic registry with raised retry attempts; DLQ consumer starts first and stops last.Reconciliation is idempotent: already-terminal requests are skipped, and optimistic locking lets a late legitimate pipeline transition win. The design mirrors SubmitQueue's proven
submitqueue/orchestrator/controller/dlqpattern, simplified for stovepipe's single payload shape.Only
processis a queue-consuming step today (ingestis synchronous RPC); future stages (build,buildsignal,record) can reuse the same pattern.Test plan
stovepipe/controller/dlq/dlq_test.go: accepted→failed, processing→failed with slot release, already-terminal no-ops, not-found no-op, CAS version-mismatch paths, malformed/empty payloadsgo build ./...andgo test ./stovepipe/... ./service/stovepipe/...passmake gazelle,make fmt,make lint-licenseclean🤖 Generated with Claude Code