feat(consumer): broker-level backoff retry middleware#19
Conversation
RetryMiddleware retries in-process, holding the handler goroutine and its prefetch slot for the whole delay, so long backoff starves the consumer. Add BackoffRetryMiddleware, which retries at the broker instead: on failure it re-publishes a delayed copy of the message back to the work queue (via the v0.4.0 delay mechanism) and acks the original, freeing the goroutine and slot during the backoff. The delay grows exponentially, snapped to the delay ladder. Attempt count is tracked in an x-rabbitwrap-retry-count header. Once the retries are exhausted the message is terminal — rejected without requeue (dead-lettered if a DLX is configured, else discarded) regardless of RequeueOnError or a handler ErrRequeue — so a failing message can never loop. A handler ErrDrop opts out of retrying immediately. Supporting additions: - Publisher.PublishDelayedToExchange: arbitrary-destination form of PublishDelayed (which now delegates to it). - DelayedPublisher interface (satisfied by *Publisher) so the middleware is testable with a fake and callers can supply any publisher. - internal Message.clone for a headers-independent retry copy. Verified with -race unit tests and the full integration suite on RabbitMQ 4 and 3.
|
Warning Review limit reached
Next review available in: 43 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 (4)
WalkthroughAdds a broker-level ChangesBroker-level backoff retry feature
Estimated code review effort: 3 (Moderate) | ~30 minutes Sequence Diagram(s)sequenceDiagram
participant Consumer
participant BackoffRetryMiddleware
participant DelayedPublisher
participant Broker
Consumer->>BackoffRetryMiddleware: handle(delivery)
BackoffRetryMiddleware->>BackoffRetryMiddleware: handler(delivery) returns error
alt retryCount < maxRetries and not ErrDrop
BackoffRetryMiddleware->>BackoffRetryMiddleware: clone delivery, bump retry-count header
BackoffRetryMiddleware->>DelayedPublisher: PublishDelayedToExchange(clonedMsg, backoffDelay)
DelayedPublisher->>Broker: schedule delayed re-publish
BackoffRetryMiddleware-->>Consumer: return nil (ack original)
else exhausted or ErrDrop
BackoffRetryMiddleware-->>Consumer: return ErrDrop-wrapped error
end
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 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❌ Patch coverage is
📢 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: 7
🤖 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 `@integration_test.go`:
- Around line 2377-2385: The detached goroutine around consumer.Consume in the
test is swallowing errors and is never synchronized with the test flow. Update
the Consume calls in this scenario to send their returned error to a buffered
channel, wrap any unexpected failure with queue-specific context, and wait for
the result after cancel() so the test observes early exits and does not leak
unsupervised goroutines.
- Around line 2404-2408: The test in integration_test.go is using a fixed 700ms
wall-clock upper bound for the slot-freed proof, which makes it flaky on slower
RabbitMQ/CI runs. Remove the timing assertion from this check and rely on the
existing [A B A] ordering in the same test to verify that B is processed before
A’s retry; keep the surrounding logic in the integration test that records got[
] and compares processing order.
- Around line 2360-2365: The test setup in DefaultConsumerConfig/WithConcurrency
is not actually proving broker-slot release because PrefetchCount still allows
extra credit; update the integration test to explicitly set prefetch to 1 in the
consumer config and keep the BackoffRetryMiddleware scenario so the second
delivery can only arrive if the failed message truly frees the broker slot. Use
the existing DefaultConsumerConfig, WithConcurrency, and WithMiddleware chain in
this test case to make the QoS behavior unambiguous.
In `@middleware.go`:
- Around line 186-192: Clamp the parsed retry count in the retry-count
extraction logic so negative values from retryCountHeader are treated as 0 (or
otherwise normalized) before being used by the retry decision path. Update the
helper that reads d.Headers[retryCountHeader] in middleware.go so the switch on
v handles int, int32, and int64 by returning a non-negative value only, and
ensure the attempt >= maxRetries check cannot be bypassed by a header like -1.
- Around line 169-172: The retry scheduling path in middleware.go is swallowing
the failure from PublishDelayedToExchange, so callers only get the original
handler error. Update the retry branch around PublishDelayedToExchange to return
or wrap schedErr with context when scheduling the retry fails, instead of always
returning err, so the failure clearly identifies the delayed
publish/queue-declare problem.
- Around line 157-168: Guard the retry path in retryCount and the retry block
that calls d.clone() against nil or zero-value Delivery inputs, since retryCount
can still panic on a nil Delivery and clone can panic when Message is missing.
Before reading headers or cloning, check that d and d.Message are valid and, if
not, return the original error unchanged instead of entering the retry logic.
In `@README.md`:
- Around line 399-405: Update the BackoffRetryMiddleware description to state
the terminal behavior explicitly: after maxRetries the middleware rejects
without requeue, and if no dead-letter exchange is configured the message is
discarded. Adjust the wording in the README text that describes the retry
exhaustion path so it no longer implies a normal consumer error fallback, and
keep the reference to BackoffRetryMiddleware and maxRetries clear for easy
location.
🪄 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: 901eed32-6d1d-44d2-9df2-d3efe84aeea9
📒 Files selected for processing (7)
CHANGELOG.mdREADME.mdintegration_test.gomessage.gomiddleware.gomiddleware_test.gopublisher.go
| go func() { | ||
| _ = consumer.Consume(ctx, func(_ context.Context, d *Delivery) error { | ||
| events <- timedEvent{string(d.Body), time.Now()} | ||
| if string(d.Body) == "A" && aAttempts.Add(1) < 2 { | ||
| return errors.New("A transient failure") // fails once, then succeeds on retry | ||
| } | ||
| return nil | ||
| }) | ||
| }() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Do not detach Consume goroutines without checking their result.
Both goroutines discard consumer.Consume errors and are never joined. If consumption exits early, the test times out with less context; if it exits during cleanup, the goroutine lifetime is unsynchronized. Send the result to a buffered channel, wrap unexpected errors with queue context, and wait for it after cancel().
As per coding guidelines, **/*.go: wrap errors with context and never silently swallow them. As per path instructions, **/*.go: scrutinize goroutine lifecycle/leaks.
Also applies to: 2445-2450
🤖 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 `@integration_test.go` around lines 2377 - 2385, The detached goroutine around
consumer.Consume in the test is swallowing errors and is never synchronized with
the test flow. Update the Consume calls in this scenario to send their returned
error to a buffered channel, wrap any unexpected failure with queue-specific
context, and wait for the result after cancel() so the test observes early exits
and does not leak unsupervised goroutines.
Sources: Coding guidelines, Path instructions
| attempt := retryCount(d) | ||
| if attempt >= maxRetries { | ||
| // Bounded: an exhausted message is terminal and must never loop, so | ||
| // force no-requeue regardless of a handler ErrRequeue or the | ||
| // consumer's RequeueOnError. Wrapping ErrDrop (and keeping the | ||
| // original only as text) makes requeueDecision reject it, so it is | ||
| // dead-lettered if a DLX is configured, else discarded. | ||
| return fmt.Errorf("rabbitmq: backoff retries (%d) exhausted: %v: %w", maxRetries, err, ErrDrop) | ||
| } | ||
|
|
||
| retryMsg := d.clone() | ||
| retryMsg.Headers[retryCountHeader] = attempt + 1 |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map relevant file(s) before reading
git ls-files | rg '^middleware\.go$|/middleware\.go$|clone|retryCount|ErrDrop'
# Read the relevant section of middleware.go
if [ -f middleware.go ]; then
sed -n '1,260p' middleware.go | cat -n
fiRepository: KARTIKrocks/rabbitwrap
Length of output: 9424
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Find the definitions used by the comment
rg -n "func .*clone|type Delivery|retryCountHeader|func retryCount|ErrDrop|RequeueOnError|clone\(" -S .Repository: KARTIKrocks/rabbitwrap
Length of output: 6369
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Focus on the exact line range around the reported snippet if middleware.go exists
if [ -f middleware.go ]; then
sed -n '130,190p' middleware.go | cat -n
fiRepository: KARTIKrocks/rabbitwrap
Length of output: 3050
🏁 Script executed:
#!/bin/bash
set -euo pipefail
git ls-files | rg '^middleware\.go$|/middleware\.go$|clone|retryCount|ErrDrop'
if [ -f middleware.go ]; then
sed -n '130,190p' middleware.go | cat -n
fiRepository: KARTIKrocks/rabbitwrap
Length of output: 3064
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n "func .*clone|type Delivery|retryCountHeader|func retryCount|ErrDrop|RequeueOnError|clone\(" -S .Repository: KARTIKrocks/rabbitwrap
Length of output: 6369
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '160,240p' message.go | cat -nRepository: KARTIKrocks/rabbitwrap
Length of output: 2729
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n "type MessageHandler|BackoffRetryMiddleware|RetryMiddleware|LoggingMiddleware|RecoveryMiddleware|handler\(ctx, d\)|next\(ctx, d\)" -S consumer.go middleware.go message.goRepository: KARTIKrocks/rabbitwrap
Length of output: 1410
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1,120p' message.go | cat -n
sed -n '560,640p' consumer.go | cat -nRepository: KARTIKrocks/rabbitwrap
Length of output: 5963
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '160,240p' message.go | cat -n
sed -n '1,120p' message.go | cat -n
sed -n '560,640p' consumer.go | cat -nRepository: KARTIKrocks/rabbitwrap
Length of output: 8532
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '160,240p' message.go | cat -n
sed -n '560,640p' consumer.go | cat -nRepository: KARTIKrocks/rabbitwrap
Length of output: 5183
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n "type MessageHandler|type Delivery struct|Message struct|zero value|nil delivery|nil message|handler receives" -S message.go consumer.go README.md middleware.goRepository: KARTIKrocks/rabbitwrap
Length of output: 411
Guard nil deliveries before cloning. retryCount still panics on a nil Delivery, and d.clone() panics on a zero-value Delivery with no Message; return the original error instead of crashing the retry path.
🤖 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 `@middleware.go` around lines 157 - 168, Guard the retry path in retryCount and
the retry block that calls d.clone() against nil or zero-value Delivery inputs,
since retryCount can still panic on a nil Delivery and clone can panic when
Message is missing. Before reading headers or cloning, check that d and
d.Message are valid and, if not, return the original error unchanged instead of
entering the retry logic.
- retryCount: clamp negative/malformed header values to 0 so a crafted x-rabbitwrap-retry-count (e.g. -1) can't bypass the attempt >= maxRetries exhaustion check. - On retry-scheduling failure, surface the PublishDelayedToExchange error via %v while preserving the original error's disposition in the chain (%w), so a broker/queue-declare problem is no longer invisible. - Integration slot-freed test: set prefetch to 1 so B can only be delivered once the failed A frees its broker slot, making [A B A] ordering the proof; drop the wall-clock upper bound on B (flaky on slow CI), keeping the robust retry-delay lower bound. - README: state the exhaustion path explicitly (rejected without requeue; discarded if no DLX; independent of RequeueOnError/ErrRequeue). Re-verified with -race unit tests and the backoff + full integration suites on RabbitMQ 4 and 3 (redelivery test run 5x for flakiness).
2 participants
RetryMiddleware retries in-process, holding the handler goroutine and its prefetch slot for the whole delay, so long backoff starves the consumer.
Add BackoffRetryMiddleware, which retries at the broker instead: on failure it re-publishes a delayed copy of the message back to the work queue (via the v0.4.0 delay mechanism) and acks the original, freeing the goroutine and slot during the backoff. The delay grows exponentially, snapped to the delay ladder. Attempt count is tracked in an x-rabbitwrap-retry-count header.
Once the retries are exhausted the message is terminal — rejected without requeue (dead-lettered if a DLX is configured, else discarded) regardless of RequeueOnError or a handler ErrRequeue — so a failing message can never loop. A handler ErrDrop opts out of retrying immediately.
Supporting additions:
Verified with -race unit tests and the full integration suite on RabbitMQ 4 and 3.
Summary
Motivation
Fixes #
Changes
Checklist
make all)Summary by CodeRabbit
New Features
Documentation
Bug Fixes