feat(stovepipe): add DLQ reconciliation for the process stage#347
feat(stovepipe): add DLQ reconciliation for the process stage#347gsturges wants to merge 2 commits into
Conversation
gsturges
requested review from
a team,
behinddwalls and
sbalabanov
as code owners
| // DLQ.TopicSuffix) when the process controller exhausts retries; DLQ.Enabled is false on its own | ||
| // subscription so a reconciliation failure retries in place rather than cascading to a further DLQ. | ||
| func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) { | ||
| dlqSub := extqueue.DefaultSubscriptionConfig(subscriberName, "stovepipe-process-dlq") |
There was a problem hiding this comment.
add an extqueue.DLQSubscriptionConfig helper that sets these values then you can just call it in the Subscription key similar to the DefaultSubscriptionConfig
| // terminal state. If the request had reached RequestStateProcessing — meaning process's | ||
| // admit step already CAS-incremented the queue's in_flight_count for it — the queue's | ||
| // slot is released first, so a crash between the two writes leaves the count still | ||
| // bound to a non-terminal request rather than double-released. |
There was a problem hiding this comment.
Is this true? Current ordering is: 1) release slot 2) potential crash 3) request state never gets updated out of processing, and could cause this whole logic to run again. In that cause it could get double released.
Either way I think it's ok for now, I don't foresee this being a typical issue and better to leave the queue unblocked than potentially deadlock it.
There was a problem hiding this comment.
You are correct, I updated comment
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
|
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 |
2 participants
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