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..4a76b24 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,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 diff --git a/integration_test.go b/integration_test.go index c0f0e9f..9e16128 100644 --- a/integration_test.go +++ b/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) + } +} 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..e416549 100644 --- a/middleware.go +++ b/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 +} diff --git a/middleware_test.go b/middleware_test.go index a8f44fd..0b85ec0 100644 --- a/middleware_test.go +++ b/middleware_test.go @@ -215,6 +215,218 @@ 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}, + {"negative clamps to 0", -1, 0}, // must not bypass attempt >= maxRetries + } + 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 {