From 81b9cd8f6c94bdad6eeeb0609775ef60902e0679 Mon Sep 17 00:00:00 2001 From: kartik Date: Sun, 5 Jul 2026 13:01:53 +0530 Subject: [PATCH 1/2] feat(consumer): broker-level backoff retry middleware MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- CHANGELOG.md | 26 ++++++ README.md | 38 ++++++-- integration_test.go | 160 +++++++++++++++++++++++++++++++++ message.go | 12 +++ middleware.go | 108 +++++++++++++++++++++++ middleware_test.go | 211 ++++++++++++++++++++++++++++++++++++++++++++ publisher.go | 17 +++- 7 files changed, 563 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 723bd94..a01ebf2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,32 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.8.0] - 2026-07-05 + +### Added + +- **Broker-level backoff retry.** `BackoffRetryMiddleware(pub, queue, maxRetries, + base)` retries a failed message at the broker instead of in-process: 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, so the handler goroutine and its + prefetch slot are freed for the whole backoff — unlike `RetryMiddleware`, which + blocks both while it sleeps. The delay grows exponentially (`base`, `2*base`, + `4*base`, …, snapped up to a `DelayLadder` rung and capped at the largest). Once + `maxRetries` scheduled retries are exhausted, the message is terminal — rejected + without requeue (dead-lettered if a dead-letter exchange is configured, else + discarded), regardless of `RequeueOnError` or a handler `ErrRequeue`, so a + failing message can never loop forever; a handler returning `ErrDrop` opts out of + retrying immediately. + + This completes the retry story started in v0.4.0/v0.6.0: prefer this over + `RetryMiddleware` for anything but short retries. Retrying is at-least-once + (scheduling the copy and acking the original are not atomic), so handlers should + be idempotent. +- `Publisher.PublishDelayedToExchange` — the arbitrary-destination form of + `PublishDelayed` (which now delegates to it). `DelayedPublisher` interface + (satisfied by `*Publisher`) captures the publish capability the new middleware + needs, so it can be supplied with any publisher. + ## [0.7.0] - 2026-07-05 ### Added diff --git a/README.md b/README.md index 6b9a94d..48b78d2 100644 --- a/README.md +++ b/README.md @@ -329,11 +329,12 @@ consConfig := rabbitmq.DefaultConsumerConfig(). #### Built-in Middleware -| Middleware | Description | -| ------------------------------------ | ------------------------------------- | -| `LoggingMiddleware(logger)` | Logs message processing with duration | -| `RecoveryMiddleware(onPanic)` | Recovers from panics in handlers | -| `RetryMiddleware(maxRetries, delay)` | Retries failed message processing | +| Middleware | Description | +| ------------------------------------------------------ | ---------------------------------------------------------------- | +| `LoggingMiddleware(logger)` | Logs message processing with duration | +| `RecoveryMiddleware(onPanic)` | Recovers from panics in handlers | +| `RetryMiddleware(maxRetries, delay)` | Retries failed processing in-process (short waits) | +| `BackoffRetryMiddleware(pub, queue, maxRetries, base)` | Retries at the broker with exponential backoff, freeing the slot | #### Custom Middleware @@ -393,6 +394,33 @@ err = consumer.Consume(ctx, func(ctx context.Context, d *rabbitmq.Delivery) erro > `RequeueOnError(true)` (without returning `ErrDrop`) reintroduces an unbounded > retry loop. +#### Broker-level backoff retry + +For anything but short retries, prefer `BackoffRetryMiddleware`. Instead of +sleeping in-process, it re-publishes a delayed copy of the failed message back to +the work queue and acks the original, so the handler goroutine and prefetch slot +are **freed** for the whole backoff — one poison message can no longer stall the +consumer. The delay grows exponentially from `base` and the message is +redelivered by the broker; after `maxRetries` the error follows the normal +disposition (dead-lettered if a dead-letter exchange is configured). + +```go +pub, _ := rabbitmq.NewPublisher(conn, rabbitmq.DefaultPublisherConfig()) + +consConfig := rabbitmq.DefaultConsumerConfig(). + WithQueue("orders"). + WithDeadLetterQueue(rabbitmq.DefaultDeadLetterConfig("orders")). // exhausted retries land here + WithMiddleware( + // 1s, 2s, 4s, ... (snapped up to the delay ladder), then dead-lettered. + rabbitmq.BackoffRetryMiddleware(pub, "orders", 5, 1*time.Second), + ) +``` + +`queue` must be a named work queue (the retry is redelivered to it by name). A +handler returning `ErrDrop` opts out of retrying. Retrying is at-least-once — +re-publishing the copy and acking the original are not atomic — so handlers should +be idempotent. + ## Health Checks ```go diff --git a/integration_test.go b/integration_test.go index c0f0e9f..e4ed711 100644 --- a/integration_test.go +++ b/integration_test.go @@ -2312,3 +2312,163 @@ func TestIntegration_DeadLetterTopologyRestoredAfterReconnect(t *testing.T) { // binding, and work-queue wiring were all re-declared. publishUntilDelivered(t, ctx, pub, dlqCh, "", work, "after") } + +// --- Broker-level backoff retry --- + +// timedEvent records a delivered message body and the time it was handled. +type timedEvent struct { + body string + when time.Time +} + +// collectTimedEvents drains exactly n events, failing on timeout or ctx cancel. +func collectTimedEvents(t *testing.T, ctx context.Context, events <-chan timedEvent, n int) []timedEvent { + t.Helper() + got := make([]timedEvent, 0, n) + for len(got) < n { + select { + case e := <-events: + got = append(got, e) + case <-time.After(15 * time.Second): + t.Fatalf("timed out; got %d/%d events: %v", len(got), n, got) + case <-ctx.Done(): + t.Fatalf("context done; got %d/%d events", len(got), n) + } + } + return got +} + +// TestIntegration_BackoffRetryRedeliversViaBroker verifies that BackoffRetryMiddleware +// retries a failed message at the broker (not in-process): the failed message is +// re-delivered after the backoff, and — crucially — the handler goroutine and its +// prefetch slot are NOT held during the wait. A second message published right +// after the first failure is processed immediately (well before the retry +// arrives), which the old in-process RetryMiddleware could not do at Concurrency=1. +func TestIntegration_BackoffRetryRedeliversViaBroker(t *testing.T) { + conn := integrationConn(t) + work := uniqueQueue(t) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + pub, err := NewPublisher(conn, DefaultPublisherConfig().WithRoutingKey(work)) + if err != nil { + t.Fatalf("failed to create publisher: %v", err) + } + t.Cleanup(func() { pub.Close() }) + + // Single goroutine: if the slot were held during the backoff (as with + // in-process retry), the second message would be serialized behind it. + cfg := DefaultConsumerConfig(). + WithConcurrency(1). + WithQueueConfig(DefaultQueueConfig(work).WithDurable(true).WithExclusive(true)). + WithMiddleware(BackoffRetryMiddleware(pub, work, 3, 1*time.Second)) + consumer, err := NewConsumer(conn, cfg) + if err != nil { + t.Fatalf("failed to create consumer: %v", err) + } + t.Cleanup(func() { consumer.Close() }) + // The holding queue is durable non-exclusive; clean it up explicitly. + t.Cleanup(func() { deleteQueue(t, conn, delayQueueName("", work, time.Second)) }) + + events := make(chan timedEvent, 8) + var aAttempts atomic.Int32 + + 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 + }) + }() + + // A fails and is scheduled for a broker-level retry; B is published right after + // and must be processed while A waits out its backoff. + if err := pub.Publish(ctx, NewTextMessage("A")); err != nil { + t.Fatalf("publish A: %v", err) + } + if err := pub.Publish(ctx, NewTextMessage("B")); err != nil { + t.Fatalf("publish B: %v", err) + } + + // Expect three deliveries in order: A (fail), B (ok), A (retry ok). + got := collectTimedEvents(t, ctx, events, 3) + if got[0].body != "A" || got[1].body != "B" || got[2].body != "A" { + t.Fatalf("delivery order = [%s %s %s], want [A B A]", got[0].body, got[1].body, got[2].body) + } + if n := aAttempts.Load(); n != 2 { + t.Errorf("A handled %d times, want 2 (initial + one retry)", n) + } + // Slot-freed proof: B is processed promptly after A's failure, not after A's + // ~1s backoff, while A's retry lands only after the delay elapses. + if gap := got[1].when.Sub(got[0].when); gap > 700*time.Millisecond { + t.Errorf("B processed %s after A's failure; slot appears held during backoff", gap) + } + if gap := got[2].when.Sub(got[0].when); gap < 900*time.Millisecond { + t.Errorf("A retry arrived after %s; expected the ~1s broker backoff", gap) + } +} + +// TestIntegration_BackoffRetryExhaustedDeadLetters verifies that once the +// broker-level retries are exhausted, the final failure follows the normal +// disposition: with a dead-letter queue configured (WithDeadLetterQueue), the +// message is dead-lettered rather than retried forever. +func TestIntegration_BackoffRetryExhaustedDeadLetters(t *testing.T) { + conn := integrationConn(t) + work := uniqueQueue(t) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + + pub, err := NewPublisher(conn, DefaultPublisherConfig().WithRoutingKey(work)) + if err != nil { + t.Fatalf("failed to create publisher: %v", err) + } + t.Cleanup(func() { pub.Close() }) + + cfg := DefaultConsumerConfig(). + WithQueueConfig(DefaultQueueConfig(work).WithDurable(true).WithExclusive(true)). + WithDeadLetterQueue(DefaultDeadLetterConfig(work)). + WithMiddleware(BackoffRetryMiddleware(pub, work, 1, 1*time.Second)) + consumer, err := NewConsumer(conn, cfg) + if err != nil { + t.Fatalf("failed to create consumer: %v", err) + } + t.Cleanup(func() { consumer.Close() }) + t.Cleanup(func() { deleteExchange(t, conn, work+".dlx") }) + t.Cleanup(func() { deleteQueue(t, conn, work+".dlq") }) + t.Cleanup(func() { deleteQueue(t, conn, delayQueueName("", work, time.Second)) }) + + var attempts atomic.Int32 + go func() { + _ = consumer.Consume(ctx, func(_ context.Context, _ *Delivery) error { + attempts.Add(1) + return errors.New("always fails") + }) + }() + + // DLQ consumer on the auto-declared dead-letter queue. + dlqConsumer, err := NewConsumer(conn, DefaultConsumerConfig().WithQueue(consumer.DeadLetterQueueName())) + if err != nil { + t.Fatalf("failed to create dlq consumer: %v", err) + } + t.Cleanup(func() { dlqConsumer.Close() }) + dlqCh, err := dlqConsumer.Start(ctx) + if err != nil { + t.Fatalf("failed to start dlq consumer: %v", err) + } + + if err := pub.Publish(ctx, NewTextMessage("poison")); err != nil { + t.Fatalf("publish: %v", err) + } + + // One broker-level retry, then exhaustion → dead-lettered to the DLQ. + recvDelivery(t, ctx, dlqCh, "poison") + + // Initial delivery + exactly one retry = 2 handler runs. + if n := attempts.Load(); n != 2 { + t.Errorf("handler ran %d times, want 2 (initial + 1 retry)", n) + } +} diff --git a/message.go b/message.go index 28e4969..0ee3e06 100644 --- a/message.go +++ b/message.go @@ -173,6 +173,18 @@ func (m *Message) WithHeaders(headers map[string]any) *Message { return m } +// clone returns a copy of the message with an independent Headers map, so the +// copy can be mutated (e.g. to stamp a retry-count header) without affecting the +// original — important when the original is a live *Delivery being redelivered. +func (m *Message) clone() *Message { + c := *m + c.Headers = make(map[string]any, len(m.Headers)) + for k, v := range m.Headers { + c.Headers[k] = v + } + return &c +} + // toPublishing converts to amqp.Publishing. func (m *Message) toPublishing() amqp.Publishing { return amqp.Publishing{ diff --git a/middleware.go b/middleware.go index 952b301..a8157a0 100644 --- a/middleware.go +++ b/middleware.go @@ -2,6 +2,7 @@ package rabbitmq import ( "context" + "errors" "fmt" "time" ) @@ -104,3 +105,110 @@ func RetryMiddleware(maxRetries int, delay time.Duration) Middleware { } } } + +// DelayedPublisher is the publish capability BackoffRetryMiddleware needs to +// schedule a retry at the broker. It is satisfied by *Publisher. +type DelayedPublisher interface { + PublishDelayedToExchange(ctx context.Context, exchange, routingKey string, msg *Message, delay time.Duration) error +} + +// retryCountHeader carries the number of broker-level retries a message has +// already been through (see BackoffRetryMiddleware). +const retryCountHeader = "x-rabbitwrap-retry-count" + +// BackoffRetryMiddleware retries a failed message at the broker instead of +// in-process. On failure it schedules a delayed copy of the message back onto +// queue (via PublishDelayedToExchange) and returns nil, so the consumer acks and +// removes the original — freeing the handler goroutine and its prefetch slot for +// the whole backoff. This is the counterpart to RetryMiddleware, which holds both +// while it sleeps; prefer this one for anything but short retries. +// +// The delay grows exponentially: base, 2*base, 4*base, ..., snapped up to the +// nearest DelayLadder rung and capped at the largest rung. After maxRetries +// scheduled retries the message is terminal: it is rejected without requeue — +// dead-lettered if a dead-letter exchange is configured, else discarded — +// regardless of the consumer's RequeueOnError or a handler ErrRequeue, so a +// failing message can never loop forever. A handler that returns ErrDrop opts out +// of retrying (the error is terminal immediately); every other non-nil error — +// including ErrRequeue — is retried until the retries are exhausted. +// +// queue must be a named work queue reachable via the default exchange (the retry +// is dead-lettered straight to it by name); server-named queues are unsupported. +// Place this middleware outermost in the chain so it observes the final error. +// A nil pub or empty queue disables retrying (errors pass through unchanged). +// +// Retrying is at-least-once: scheduling the copy and acking the original are not +// atomic, so a crash in between can duplicate the message. Handlers should be +// idempotent. +func BackoffRetryMiddleware(pub DelayedPublisher, queue string, maxRetries int, base time.Duration) Middleware { + if maxRetries < 0 { + maxRetries = 0 + } + return func(next MessageHandler) MessageHandler { + return func(ctx context.Context, d *Delivery) error { + err := next(ctx, d) + if err == nil { + return nil + } + if pub == nil || queue == "" || errors.Is(err, ErrDrop) { + return err + } + + 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 + if schedErr := pub.PublishDelayedToExchange(ctx, "", queue, retryMsg, backoffDelay(base, attempt)); schedErr != nil { + // Couldn't schedule the retry — fall back to the normal + // disposition of the original failure. + return err + } + return nil + } + } +} + +// retryCount reads the broker-level retry count from a delivery's headers. AMQP +// round-trips integer headers as int32/int64, so several integer types are +// accepted; a missing or unrecognized header counts as zero. +func retryCount(d *Delivery) int { + if d.Message == nil { + return 0 + } + switch v := d.Headers[retryCountHeader].(type) { + case int: + return v + case int32: + return int(v) + case int64: + return int(v) + default: + return 0 + } +} + +// backoffDelay returns the delay before retry number attempt (0-based): +// base * 2^attempt, with base<=0 defaulting to 1s and the result clamped to the +// largest DelayLadder rung so it never overflows or exceeds ErrDelayTooLong. +func backoffDelay(base time.Duration, attempt int) time.Duration { + if base <= 0 { + base = 1 * time.Second + } + maxDelay := delayLadder[len(delayLadder)-1] + delay := base + for i := 0; i < attempt && delay < maxDelay; i++ { + delay *= 2 + } + if delay > maxDelay { + delay = maxDelay + } + return delay +} diff --git a/middleware_test.go b/middleware_test.go index a8f44fd..4947792 100644 --- a/middleware_test.go +++ b/middleware_test.go @@ -215,6 +215,217 @@ func TestLoggingMiddlewareNilLogger(t *testing.T) { } } +// fakeDelayedPublisher records the arguments of the last (and count of all) +// PublishDelayedToExchange calls so BackoffRetryMiddleware can be unit-tested +// without a broker. +type fakeDelayedPublisher struct { + calls int + exchange string + routingKey string + msg *Message + delay time.Duration + err error // returned by PublishDelayedToExchange +} + +func (f *fakeDelayedPublisher) PublishDelayedToExchange(_ context.Context, exchange, routingKey string, msg *Message, delay time.Duration) error { + f.calls++ + f.exchange = exchange + f.routingKey = routingKey + f.msg = msg + f.delay = delay + return f.err +} + +func TestBackoffDelay(t *testing.T) { + maxDelay := delayLadder[len(delayLadder)-1] + + tests := []struct { + base time.Duration + attempt int + want time.Duration + }{ + {1 * time.Second, 0, 1 * time.Second}, + {1 * time.Second, 1, 2 * time.Second}, + {1 * time.Second, 3, 8 * time.Second}, + {2 * time.Second, 2, 8 * time.Second}, + {0, 0, 1 * time.Second}, // base <= 0 defaults to 1s + {-5 * time.Second, 1, 2 * time.Second}, // negative base defaults to 1s + {1 * time.Second, 100, maxDelay}, // huge attempt clamps to the max rung + } + for _, tt := range tests { + if got := backoffDelay(tt.base, tt.attempt); got != tt.want { + t.Errorf("backoffDelay(%s, %d) = %s, want %s", tt.base, tt.attempt, got, tt.want) + } + } +} + +func TestRetryCount(t *testing.T) { + tests := []struct { + name string + value any + want int + }{ + {"missing", nil, 0}, + {"int", 3, 3}, + {"int32", int32(4), 4}, + {"int64", int64(5), 5}, + {"unrecognized", "nope", 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + headers := map[string]any{} + if tt.value != nil { + headers[retryCountHeader] = tt.value + } + d := &Delivery{Message: &Message{Headers: headers}} + if got := retryCount(d); got != tt.want { + t.Errorf("retryCount = %d, want %d", got, tt.want) + } + }) + } +} + +func TestBackoffRetryMiddlewareSuccess(t *testing.T) { + pub := &fakeDelayedPublisher{} + handler := BackoffRetryMiddleware(pub, "work", 3, time.Second)(func(_ context.Context, _ *Delivery) error { + return nil + }) + + if err := handler(context.Background(), &Delivery{Message: &Message{}}); err != nil { + t.Errorf("expected nil, got %v", err) + } + if pub.calls != 0 { + t.Errorf("expected no publish on success, got %d", pub.calls) + } +} + +func TestBackoffRetryMiddlewareSchedulesRetry(t *testing.T) { + pub := &fakeDelayedPublisher{} + wantErr := errors.New("boom") + handler := BackoffRetryMiddleware(pub, "work", 3, time.Second)(func(_ context.Context, _ *Delivery) error { + return wantErr + }) + + // First delivery: no retry-count header -> attempt 0. + d := &Delivery{Message: &Message{Headers: map[string]any{}}} + err := handler(context.Background(), d) + if err != nil { + t.Fatalf("expected nil (retry scheduled), got %v", err) + } + if pub.calls != 1 { + t.Fatalf("expected 1 publish, got %d", pub.calls) + } + if pub.exchange != "" || pub.routingKey != "work" { + t.Errorf("expected redelivery to (\"\", work), got (%q, %q)", pub.exchange, pub.routingKey) + } + if pub.delay != time.Second { + t.Errorf("expected delay 1s for attempt 0, got %s", pub.delay) + } + if got := pub.msg.Headers[retryCountHeader]; got != 1 { // attempt 0 + 1 + t.Errorf("expected retry-count header 1 on the copy, got %v", got) + } + // The original delivery must not be mutated by the copy. + if _, ok := d.Headers[retryCountHeader]; ok { + t.Error("original delivery headers were mutated") + } +} + +func TestBackoffRetryMiddlewareExhausted(t *testing.T) { + pub := &fakeDelayedPublisher{} + handler := BackoffRetryMiddleware(pub, "work", 2, time.Second)(func(_ context.Context, _ *Delivery) error { + return errors.New("boom") + }) + + // Seed the header so this is already the max-th retry. + d := &Delivery{Message: &Message{Headers: map[string]any{retryCountHeader: 2}}} + err := handler(context.Background(), d) + if pub.calls != 0 { + t.Errorf("expected no publish once exhausted, got %d", pub.calls) + } + // Exhausted retries are terminal: forced no-requeue even if the consumer + // defaults to requeue, so a failing message can never loop. + if !errors.Is(err, ErrDrop) { + t.Errorf("expected exhausted error to force ErrDrop, got %v", err) + } + if requeueDecision(err, true) { + t.Error("exhausted retry must not requeue even with RequeueOnError=true") + } +} + +// TestBackoffRetryMiddlewareExhaustedNeutralizesRequeue ensures a handler that +// keeps returning ErrRequeue cannot cause an unbounded requeue loop: once the +// retries are exhausted the message is forced terminal. +func TestBackoffRetryMiddlewareExhaustedNeutralizesRequeue(t *testing.T) { + pub := &fakeDelayedPublisher{} + handler := BackoffRetryMiddleware(pub, "work", 1, time.Second)(func(_ context.Context, _ *Delivery) error { + return fmt.Errorf("transient: %w", ErrRequeue) + }) + + d := &Delivery{Message: &Message{Headers: map[string]any{retryCountHeader: 1}}} + err := handler(context.Background(), d) + if pub.calls != 0 { + t.Errorf("expected no publish once exhausted, got %d", pub.calls) + } + if requeueDecision(err, false) { + t.Error("ErrRequeue at exhaustion must not requeue (would loop forever)") + } +} + +func TestBackoffRetryMiddlewareDropIsTerminal(t *testing.T) { + pub := &fakeDelayedPublisher{} + handler := BackoffRetryMiddleware(pub, "work", 3, time.Second)(func(_ context.Context, _ *Delivery) error { + return fmt.Errorf("poison: %w", ErrDrop) + }) + + err := handler(context.Background(), &Delivery{Message: &Message{Headers: map[string]any{}}}) + if !errors.Is(err, ErrDrop) { + t.Errorf("expected ErrDrop to pass through, got %v", err) + } + if pub.calls != 0 { + t.Errorf("expected no publish for ErrDrop, got %d", pub.calls) + } +} + +func TestBackoffRetryMiddlewareScheduleFailureFallsBack(t *testing.T) { + wantErr := errors.New("boom") + pub := &fakeDelayedPublisher{err: errors.New("broker down")} + handler := BackoffRetryMiddleware(pub, "work", 3, time.Second)(func(_ context.Context, _ *Delivery) error { + return wantErr + }) + + err := handler(context.Background(), &Delivery{Message: &Message{Headers: map[string]any{}}}) + if !errors.Is(err, wantErr) { + t.Errorf("expected original error when scheduling fails, got %v", err) + } + if pub.calls != 1 { + t.Errorf("expected one publish attempt, got %d", pub.calls) + } +} + +func TestBackoffRetryMiddlewareDisabled(t *testing.T) { + wantErr := errors.New("boom") + + // nil publisher and empty queue both disable retrying without panicking. + for _, tc := range []struct { + name string + pub DelayedPublisher + queue string + }{ + {"nil publisher", nil, "work"}, + {"empty queue", &fakeDelayedPublisher{}, ""}, + } { + t.Run(tc.name, func(t *testing.T) { + handler := BackoffRetryMiddleware(tc.pub, tc.queue, 3, time.Second)(func(_ context.Context, _ *Delivery) error { + return wantErr + }) + err := handler(context.Background(), &Delivery{Message: &Message{Headers: map[string]any{}}}) + if !errors.Is(err, wantErr) { + t.Errorf("expected error to pass through, got %v", err) + } + }) + } +} + // testLogger is a test helper for Logger interface. type testLogger struct { onDebugf func(string, ...any) diff --git a/publisher.go b/publisher.go index de60619..51bb1f6 100644 --- a/publisher.go +++ b/publisher.go @@ -335,7 +335,8 @@ func delayQueueName(exchange, routingKey string, delay time.Duration) string { } // PublishDelayed publishes a message that is delivered to the publisher's -// configured exchange and routing key only after the given delay. +// configured exchange and routing key only after the given delay. It is a thin +// wrapper over PublishDelayedToExchange targeting the configured destination. // // It works without any broker plugin: the message is published into a dedicated // holding queue whose queue-level TTL equals the delay and whose dead-letter @@ -353,11 +354,21 @@ func delayQueueName(exchange, routingKey string, delay time.Duration) string { // not for precise scheduling. Dead-lettering does not honor the Mandatory flag, // so a message routed to a destination with no queue is dropped on expiry. func (p *Publisher) PublishDelayed(ctx context.Context, msg *Message, delay time.Duration) error { + return p.PublishDelayedToExchange(ctx, p.config.Exchange, p.config.RoutingKey, msg, delay) +} + +// PublishDelayedToExchange publishes a message that is delivered to the given +// exchange and routing key only after the given delay. It is the arbitrary +// destination form of PublishDelayed (which targets the publisher's configured +// exchange/routing key); see PublishDelayed for the full mechanism and timing +// semantics. Each distinct (exchange, routingKey, delay) uses its own holding +// queue, so different destinations never share one. +func (p *Publisher) PublishDelayedToExchange(ctx context.Context, exchange, routingKey string, msg *Message, delay time.Duration) error { if msg == nil { return ErrNilMessage } if delay <= 0 { - return p.Publish(ctx, msg) + return p.PublishToExchange(ctx, exchange, routingKey, msg) } snapped, err := snapDelay(delay) @@ -365,8 +376,6 @@ func (p *Publisher) PublishDelayed(ctx context.Context, msg *Message, delay time return err } - exchange := p.config.Exchange - routingKey := p.config.RoutingKey queueName := delayQueueName(exchange, routingKey, snapped) if err := p.declareDelayQueue(queueName, exchange, routingKey, snapped); err != nil { From 4bc009dcc341e73462d389ba35083fe3f4988f96 Mon Sep 17 00:00:00 2001 From: kartik Date: Sun, 5 Jul 2026 13:18:15 +0530 Subject: [PATCH 2/2] refactor(consumer): address review feedback on backoff retry - 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). --- README.md | 6 ++++-- integration_test.go | 18 ++++++++++-------- middleware.go | 22 ++++++++++++++-------- middleware_test.go | 1 + 4 files changed, 29 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index 48b78d2..4a76b24 100644 --- a/README.md +++ b/README.md @@ -401,8 +401,10 @@ sleeping in-process, it re-publishes a delayed copy of the failed message back t the work queue and acks the original, so the handler goroutine and prefetch slot are **freed** for the whole backoff — one poison message can no longer stall the consumer. The delay grows exponentially from `base` and the message is -redelivered by the broker; after `maxRetries` the error follows the normal -disposition (dead-lettered if a dead-letter exchange is configured). +redelivered by the broker. After `maxRetries` the message is terminal: it is +rejected **without requeue** — dead-lettered if a dead-letter exchange is +configured, otherwise discarded — regardless of `RequeueOnError` or a handler +`ErrRequeue`, so it can never loop forever. ```go pub, _ := rabbitmq.NewPublisher(conn, rabbitmq.DefaultPublisherConfig()) diff --git a/integration_test.go b/integration_test.go index e4ed711..9e16128 100644 --- a/integration_test.go +++ b/integration_test.go @@ -2357,10 +2357,13 @@ func TestIntegration_BackoffRetryRedeliversViaBroker(t *testing.T) { } t.Cleanup(func() { pub.Close() }) - // Single goroutine: if the slot were held during the backoff (as with - // in-process retry), the second message would be serialized behind it. + // Prefetch 1 + a single goroutine: the broker sends the next message only + // after the current one is acked, so B can arrive only if the failed A truly + // frees its slot (the middleware acks A after scheduling the broker retry). An + // in-process retry would hold A unacked and B would never arrive. cfg := DefaultConsumerConfig(). WithConcurrency(1). + WithPrefetch(1, 0). WithQueueConfig(DefaultQueueConfig(work).WithDurable(true).WithExclusive(true)). WithMiddleware(BackoffRetryMiddleware(pub, work, 3, 1*time.Second)) consumer, err := NewConsumer(conn, cfg) @@ -2393,7 +2396,10 @@ func TestIntegration_BackoffRetryRedeliversViaBroker(t *testing.T) { t.Fatalf("publish B: %v", err) } - // Expect three deliveries in order: A (fail), B (ok), A (retry ok). + // Expect three deliveries in order: A (fail), B (ok), A (retry ok). Under + // prefetch 1 this ordering is itself the slot-freed proof: B can only be + // delivered once A's slot is released, and it arrives before A's delayed + // retry. (No wall-clock upper bound on B — that would be flaky on slow CI.) got := collectTimedEvents(t, ctx, events, 3) if got[0].body != "A" || got[1].body != "B" || got[2].body != "A" { t.Fatalf("delivery order = [%s %s %s], want [A B A]", got[0].body, got[1].body, got[2].body) @@ -2401,11 +2407,7 @@ func TestIntegration_BackoffRetryRedeliversViaBroker(t *testing.T) { if n := aAttempts.Load(); n != 2 { t.Errorf("A handled %d times, want 2 (initial + one retry)", n) } - // Slot-freed proof: B is processed promptly after A's failure, not after A's - // ~1s backoff, while A's retry lands only after the delay elapses. - if gap := got[1].when.Sub(got[0].when); gap > 700*time.Millisecond { - t.Errorf("B processed %s after A's failure; slot appears held during backoff", gap) - } + // The retry is genuinely broker-delayed, not an instant redelivery. if gap := got[2].when.Sub(got[0].when); gap < 900*time.Millisecond { t.Errorf("A retry arrived after %s; expected the ~1s broker backoff", gap) } diff --git a/middleware.go b/middleware.go index a8157a0..e416549 100644 --- a/middleware.go +++ b/middleware.go @@ -167,9 +167,11 @@ func BackoffRetryMiddleware(pub DelayedPublisher, queue string, maxRetries int, retryMsg := d.clone() retryMsg.Headers[retryCountHeader] = attempt + 1 if schedErr := pub.PublishDelayedToExchange(ctx, "", queue, retryMsg, backoffDelay(base, attempt)); schedErr != nil { - // Couldn't schedule the retry — fall back to the normal - // disposition of the original failure. - return err + // Couldn't schedule the retry (delayed publish / queue declare + // failed): surface that failure while still applying the original + // error's disposition (kept in the chain via %w, so an ErrRequeue + // still requeues and a plain error still dead-letters/discards). + return fmt.Errorf("rabbitmq: scheduling backoff retry failed: %v: %w", schedErr, err) } return nil } @@ -178,21 +180,25 @@ func BackoffRetryMiddleware(pub DelayedPublisher, queue string, maxRetries int, // retryCount reads the broker-level retry count from a delivery's headers. AMQP // round-trips integer headers as int32/int64, so several integer types are -// accepted; a missing or unrecognized header counts as zero. +// accepted; a missing, unrecognized, or negative header counts as zero (a +// negative value must not let the attempt >= maxRetries check be bypassed). func retryCount(d *Delivery) int { if d.Message == nil { return 0 } + var n int switch v := d.Headers[retryCountHeader].(type) { case int: - return v + n = v case int32: - return int(v) + n = int(v) case int64: - return int(v) - default: + n = int(v) + } + if n < 0 { return 0 } + return n } // backoffDelay returns the delay before retry number attempt (0-based): diff --git a/middleware_test.go b/middleware_test.go index 4947792..0b85ec0 100644 --- a/middleware_test.go +++ b/middleware_test.go @@ -270,6 +270,7 @@ func TestRetryCount(t *testing.T) { {"int32", int32(4), 4}, {"int64", int64(5), 5}, {"unrecognized", "nope", 0}, + {"negative clamps to 0", -1, 0}, // must not bypass attempt >= maxRetries } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) {