feat(consumer): broker-level backoff retry middleware

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

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,35 @@ 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 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())
+
+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

integration_test.go

@@ -2312,3 +2312,165 @@ 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() })
+
+	// 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)
+	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). 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)
+	}
+	if n := aAttempts.Load(); n != 2 {
+		t.Errorf("A handled %d times, want 2 (initial + one retry)", n)
+	}
+	// 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)
+	}
+}
+
+// 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)
+	}
+}

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{

middleware.go

@@ -2,6 +2,7 @@ package rabbitmq
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 )
@@ -104,3 +105,116 @@ 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 (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
+		}
+	}
+}
+
+// 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, 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:
+		n = v
+	case int32:
+		n = int(v)
+	case int64:
+		n = int(v)
+	}
+	if n < 0 {
+		return 0
+	}
+	return n
+}
+
+// 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
+}