feat(consumer): safe requeue default + per-message disposition sentinels#17
Conversation
Flip DefaultConsumerConfig().RequeueOnError from true to false. Previously a handler error requeued the message immediately and forever, bypassing any DLQ — a single poison message hot-looped and head-of-line-blocked the queue. A failed message is now rejected without requeue (dead-lettered if a DLX is configured, else discarded). Behavioral change; restore with WithRequeueOnError(true). Add ErrRequeue / ErrDrop sentinel errors: a handler can return them (optionally %w-wrapped) to force requeue or no-requeue per message, overriding the default. The decision is a pure requeueDecision() helper, wired into processDelivery. This also fixes RetryMiddleware causing unbounded redelivery under the default: after in-process retries are exhausted the error is rejected, not requeued. Its doc comment now spells out the in-process cost and the RequeueOnError interaction.
|
Warning Review limit reached
Next review available in: 38 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
WalkthroughThis PR changes ChangesRequeue behavior change
Estimated code review effort: 3 (Moderate) | ~25 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Codecov Report✅ All modified and coverable lines are covered by tests. 📢 Thoughts on this report? Let us know! |
|
@coderabbitai full review |
✅ Action performedFull review finished. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
consumer.go (1)
508-522: 📐 Maintainability & Code Quality | 🔵 TrivialCodecov flagged one uncovered line in this function — likely the
nackErrbranch (Lines 515-519).That path (
delivery.Nackitself failing) is hard to exercise without a way to inject a failingNack. Do you want me to sketch a test double forDelivery/the underlying channel to cover this branch, or open a tracking issue instead?As per the PR objectives, "Codecov reported patch coverage of 90% with 1 changed line in consumer.go missing coverage and asked for review of the uncovered line."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@consumer.go` around lines 508 - 522, The uncovered branch is the nack failure path inside Consumer.processDelivery when delivery.Nack returns an error. Add a focused unit test for processDelivery that uses a test double for Delivery (or the underlying channel wrapper) to force Nack(false, requeueDecision(...)) to fail, and assert that c.config.OnError is invoked with nackErr when AutoAck is false. Use the existing processDelivery, Delivery.Nack, and requeueDecision symbols to locate and exercise this branch.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@consumer_test.go`:
- Around line 44-66: The table-driven subtests in TestRequeueDecision are
independent because requeueDecision is pure, so mark each t.Run subtest with
t.Parallel to follow the test style guidance. Update the subtest body inside
TestRequeueDecision so each case runs in parallel without sharing state, keeping
the existing table-driven structure and assertions intact.
In `@integration_test.go`:
- Around line 2125-2131: The negative assertion in the hot-loop check uses a
fixed sleep, which makes the test flaky under slow or contended conditions.
Update the assertion in the integration test around attempts.Load() to use
bounded polling with a short interval and deadline, or compare successive reads
over a couple of polls, so it deterministically verifies the handler never
exceeds one run without relying on a single time.Sleep.
In `@middleware.go`:
- Around line 70-77: Update the godoc on RetryMiddleware to correct the
“dead-lettered” wording: clarify that after retries are exhausted the message is
only dead-lettered if the queue has a DLX configured, otherwise it is discarded,
matching the ErrDrop documentation in rabbitmq.go. Also add a note that
RetryMiddleware retries every non-nil error returned by the handler, including
ErrDrop and ErrRequeue, and that these sentinels only affect the final
disposition after maxRetries is reached, not whether retries occur.
---
Outside diff comments:
In `@consumer.go`:
- Around line 508-522: The uncovered branch is the nack failure path inside
Consumer.processDelivery when delivery.Nack returns an error. Add a focused unit
test for processDelivery that uses a test double for Delivery (or the underlying
channel wrapper) to force Nack(false, requeueDecision(...)) to fail, and assert
that c.config.OnError is invoked with nackErr when AutoAck is false. Use the
existing processDelivery, Delivery.Nack, and requeueDecision symbols to locate
and exercise this branch.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 4b6ea00f-45f4-4677-ac69-c68e39632575
📒 Files selected for processing (8)
CHANGELOG.mdREADME.mdconsumer.goconsumer_test.gointegration_test.gomiddleware.gomiddleware_test.gorabbitmq.go
| func TestRequeueDecision(t *testing.T) { | ||
| tests := []struct { | ||
| name string | ||
| err error | ||
| defaultRequeue bool | ||
| want bool | ||
| }{ | ||
| {"nil error, default false", nil, false, false}, | ||
| {"nil error, default true", nil, true, true}, | ||
| {"plain error follows default false", errors.New("boom"), false, false}, | ||
| {"plain error follows default true", errors.New("boom"), true, true}, | ||
| {"ErrRequeue overrides default false", ErrRequeue, false, true}, | ||
| {"ErrRequeue wrapped overrides default false", fmt.Errorf("db down: %w", ErrRequeue), false, true}, | ||
| {"ErrDrop overrides default true", ErrDrop, true, false}, | ||
| {"ErrDrop wrapped overrides default true", fmt.Errorf("poison: %w", ErrDrop), true, false}, | ||
| } | ||
| for _, tt := range tests { | ||
| t.Run(tt.name, func(t *testing.T) { | ||
| if got := requeueDecision(tt.err, tt.defaultRequeue); got != tt.want { | ||
| t.Errorf("requeueDecision(%v, %v) = %v, want %v", tt.err, tt.defaultRequeue, got, tt.want) | ||
| } | ||
| }) | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
Consider t.Parallel() for the table-driven subtests.
requeueDecision is a pure function with no shared state, so each t.Run subtest is safe to parallelize.
♻️ Proposed diff
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
+ t.Parallel()
if got := requeueDecision(tt.err, tt.defaultRequeue); got != tt.want {
t.Errorf("requeueDecision(%v, %v) = %v, want %v", tt.err, tt.defaultRequeue, got, tt.want)
}
})
}As per path instructions, "Prefer table-driven tests and t.Parallel where safe."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@consumer_test.go` around lines 44 - 66, The table-driven subtests in
TestRequeueDecision are independent because requeueDecision is pure, so mark
each t.Run subtest with t.Parallel to follow the test style guidance. Update the
subtest body inside TestRequeueDecision so each case runs in parallel without
sharing state, keeping the existing table-driven structure and assertions
intact.
Source: Path instructions
- Cover the previously-untested nack-failure branch in processDelivery with a focused unit test (a Delivery with a nil Acknowledger forces Nack to error), asserting OnError receives the nack error. - Replace the hot-loop check's single time.Sleep with bounded polling that fails the instant the attempt count climbs past one — deterministic, not timing-luck. - Correct RetryMiddleware godoc: after retries exhaust, the message is dead-lettered only if the queue has a DLX (else discarded), and note that every error is retried, with ErrRequeue/ErrDrop affecting only the final disposition.
2 participants
Flip DefaultConsumerConfig().RequeueOnError from true to false. Previously a handler error requeued the message immediately and forever, bypassing any DLQ — a single poison message hot-looped and head-of-line-blocked the queue. A failed message is now rejected without requeue (dead-lettered if a DLX is configured, else discarded). Behavioral change; restore with WithRequeueOnError(true).
Add ErrRequeue / ErrDrop sentinel errors: a handler can return them (optionally %w-wrapped) to force requeue or no-requeue per message, overriding the default. The decision is a pure requeueDecision() helper, wired into processDelivery.
This also fixes RetryMiddleware causing unbounded redelivery under the default: after in-process retries are exhausted the error is rejected, not requeued. Its doc comment now spells out the in-process cost and the RequeueOnError interaction.
Summary
Motivation
Fixes #
Changes
Checklist
make all)Summary by CodeRabbit
New Features
Bug Fixes
Documentation