From f7da1d606ad43c2d06bfe1b0536144106729bf98 Mon Sep 17 00:00:00 2001 From: kartik Date: Sun, 5 Jul 2026 08:34:47 +0530 Subject: [PATCH 1/2] feat(consumer): safe requeue default + per-message disposition sentinels MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Flip DefaultConsumerConfig().RequeueOnError from true to false. Previously a handler error requeued the message immediately and forever, bypassing any DLQ — a single poison message hot-looped and head-of-line-blocked the queue. A failed message is now rejected without requeue (dead-lettered if a DLX is configured, else discarded). Behavioral change; restore with WithRequeueOnError(true). Add ErrRequeue / ErrDrop sentinel errors: a handler can return them (optionally %w-wrapped) to force requeue or no-requeue per message, overriding the default. The decision is a pure requeueDecision() helper, wired into processDelivery. This also fixes RetryMiddleware causing unbounded redelivery under the default: after in-process retries are exhausted the error is rejected, not requeued. Its doc comment now spells out the in-process cost and the RequeueOnError interaction. --- CHANGELOG.md | 28 ++++++++++ README.md | 28 +++++++++- consumer.go | 25 +++++++-- consumer_test.go | 30 ++++++++++- integration_test.go | 124 ++++++++++++++++++++++++++++++++++++++++++++ middleware.go | 8 +++ middleware_test.go | 25 +++++++++ rabbitmq.go | 9 ++++ 8 files changed, 271 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ca79242..713b3f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,34 @@ 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.6.0] - 2026-07-05 + +### Changed + +- **`DefaultConsumerConfig().RequeueOnError` now defaults to `false`** (was `true`). + Previously a handler that returned an error caused the message to be requeued + immediately, forever, with no delay and bypassing any dead-letter queue — a + single poison message became a CPU-burning hot loop that head-of-line-blocked + the queue. A failed message is now rejected without requeue: dead-lettered if a + dead-letter exchange is configured, otherwise discarded. This is a behavioral + change — restore the old behavior with `WithRequeueOnError(true)`. (Mirrors the + 0.3.0 change that made `ConfirmMode` default to `false`.) + +### Added + +- **Per-message requeue control via sentinel errors.** A handler can return + `ErrRequeue` to force the message to be requeued (for transient failures) or + `ErrDrop` to force it not to be requeued (for poison messages), overriding the + configured `RequeueOnError` default. Both may be wrapped with `%w`. + +### Fixed + +- **`RetryMiddleware` no longer causes unbounded redelivery under the default + config.** After its in-process retries are exhausted the error propagates to + the consumer, which now rejects (dead-letters) the message instead of requeuing + it. Its doc comment now spells out the interaction with `RequeueOnError` and + that retries are in-process (holding the handler goroutine and prefetch slot). + ## [0.5.0] - 2026-07-05 ### Added diff --git a/README.md b/README.md index 876b7e0..0d6e8fa 100644 --- a/README.md +++ b/README.md @@ -331,15 +331,41 @@ handler := combined(myHandler) ### Error Handling +When a handler returns an error, the message is nacked. By default +(`RequeueOnError: false`) it is **not** requeued — it is dead-lettered if a +dead-letter exchange is configured, otherwise discarded. This avoids a poison +message hot-looping. Opt into unconditional requeue with `WithRequeueOnError(true)`. + ```go consConfig := rabbitmq.DefaultConsumerConfig(). WithQueue("my-queue"). - WithRequeueOnError(true). WithErrorHandler(func(err error) { log.Printf("Consumer error: %v", err) }) ``` +For per-message control, return a sentinel error from the handler — it overrides +the `RequeueOnError` default and may be wrapped with `%w`: + +```go +err = consumer.Consume(ctx, func(ctx context.Context, d *rabbitmq.Delivery) error { + if err := process(d); err != nil { + if isTransient(err) { + return fmt.Errorf("temporary: %w", rabbitmq.ErrRequeue) // requeue and retry + } + return fmt.Errorf("poison: %w", rabbitmq.ErrDrop) // never requeue (dead-letter/discard) + } + return nil +}) +``` + +> **`RetryMiddleware`**: retries happen in-process (the handler goroutine and its +> prefetch slot are held for the delay), so it suits short retries, not long +> backoff. After the retries are exhausted the error is nacked per the rules +> above — so with the default it is dead-lettered. Combining it with +> `RequeueOnError(true)` (without returning `ErrDrop`) reintroduces an unbounded +> retry loop. + ## Health Checks ```go diff --git a/consumer.go b/consumer.go index fdf40a6..ecf8853 100644 --- a/consumer.go +++ b/consumer.go @@ -2,6 +2,7 @@ package rabbitmq import ( "context" + "errors" "fmt" "maps" "sync" @@ -39,7 +40,11 @@ type ConsumerConfig struct { // PrefetchSize is the prefetch size in bytes. PrefetchSize int - // RequeueOnError requeues messages when handler returns error. + // RequeueOnError requeues a message when the handler returns an error + // (default: false). When false, a failed message is rejected without + // requeue — dead-lettered if a dead-letter exchange is configured, else + // discarded — which avoids a poison message hot-looping. A handler can + // override this per message by returning ErrRequeue or ErrDrop. RequeueOnError bool // Concurrency is the number of goroutines processing messages (default: 1). @@ -74,7 +79,7 @@ func DefaultConsumerConfig() ConsumerConfig { NoWait: false, PrefetchCount: 10, PrefetchSize: 0, - RequeueOnError: true, + RequeueOnError: false, Concurrency: 1, GracefulShutdown: true, } @@ -486,6 +491,20 @@ func (c *Consumer) Consume(ctx context.Context, handler MessageHandler) error { return <-errCh } +// requeueDecision decides whether a failed delivery is requeued. An explicit +// ErrRequeue or ErrDrop returned by the handler overrides the configured +// default; ErrRequeue wins if both are somehow present. +func requeueDecision(err error, defaultRequeue bool) bool { + switch { + case errors.Is(err, ErrRequeue): + return true + case errors.Is(err, ErrDrop): + return false + default: + return defaultRequeue + } +} + // processDelivery handles a single delivery with ack/nack logic. func (c *Consumer) processDelivery(ctx context.Context, handler MessageHandler, delivery *Delivery) { if err := handler(ctx, delivery); err != nil { @@ -493,7 +512,7 @@ func (c *Consumer) processDelivery(ctx context.Context, handler MessageHandler, c.config.OnError(err) } if !c.config.AutoAck { - if nackErr := delivery.Nack(false, c.config.RequeueOnError); nackErr != nil { + if nackErr := delivery.Nack(false, requeueDecision(err, c.config.RequeueOnError)); nackErr != nil { if c.config.OnError != nil { c.config.OnError(nackErr) } diff --git a/consumer_test.go b/consumer_test.go index 86ee3d2..83d3732 100644 --- a/consumer_test.go +++ b/consumer_test.go @@ -2,6 +2,7 @@ package rabbitmq import ( "errors" + "fmt" "testing" "time" ) @@ -35,8 +36,33 @@ func TestDefaultConsumerConfig(t *testing.T) { if c.PrefetchCount != 10 { t.Errorf("expected PrefetchCount 10, got %d", c.PrefetchCount) } - if !c.RequeueOnError { - t.Error("expected RequeueOnError true") + if c.RequeueOnError { + t.Error("expected RequeueOnError false (safe default)") + } +} + +func TestRequeueDecision(t *testing.T) { + tests := []struct { + name string + err error + defaultRequeue bool + want bool + }{ + {"nil error, default false", nil, false, false}, + {"nil error, default true", nil, true, true}, + {"plain error follows default false", errors.New("boom"), false, false}, + {"plain error follows default true", errors.New("boom"), true, true}, + {"ErrRequeue overrides default false", ErrRequeue, false, true}, + {"ErrRequeue wrapped overrides default false", fmt.Errorf("db down: %w", ErrRequeue), false, true}, + {"ErrDrop overrides default true", ErrDrop, true, false}, + {"ErrDrop wrapped overrides default true", fmt.Errorf("poison: %w", ErrDrop), true, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := requeueDecision(tt.err, tt.defaultRequeue); got != tt.want { + t.Errorf("requeueDecision(%v, %v) = %v, want %v", tt.err, tt.defaultRequeue, got, tt.want) + } + }) } } diff --git a/integration_test.go b/integration_test.go index d235133..379b41a 100644 --- a/integration_test.go +++ b/integration_test.go @@ -5,6 +5,7 @@ package rabbitmq import ( "context" "encoding/json" + "errors" "fmt" "os" "sync" @@ -2057,3 +2058,126 @@ func deleteQueue(t *testing.T, conn *Connection, queue string) { defer ch.Close() _, _ = ch.Raw().QueueDelete(queue, false, false, false) } + +// --- Requeue disposition --- + +// TestIntegration_TerminalErrorNotRequeued verifies that with the default +// config (RequeueOnError=false) a handler that always errors does NOT hot-loop: +// the message is rejected once and dead-lettered, and the handler runs a +// bounded number of times. +func TestIntegration_TerminalErrorNotRequeued(t *testing.T) { + conn := integrationConn(t) + mainQ := uniqueQueue(t) + dlx := mainQ + ".dlx" + dlq := mainQ + ".dlq" + + pub, err := NewPublisher(conn, DefaultPublisherConfig().WithRoutingKey(mainQ)) + if err != nil { + t.Fatalf("failed to create publisher: %v", err) + } + t.Cleanup(func() { pub.Close() }) + + // Dead-letter exchange (fanout so the routing key is irrelevant). + if err := pub.DeclareExchange(dlx, ExchangeFanout, false, false, nil); err != nil { + t.Fatalf("declare dlx: %v", err) + } + t.Cleanup(func() { deleteExchange(t, conn, dlx) }) + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + // DLQ consumer: declares the DLQ and binds it to the DLX. + dlqConsumer, err := NewConsumer(conn, DefaultConsumerConfig(). + WithQueueConfig(DefaultQueueConfig(dlq).WithDurable(false).WithAutoDelete(true).WithExclusive(true)). + WithBinding(dlx, "", nil)) + 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) + } + + // Main consumer: default config (RequeueOnError=false), queue dead-letters to dlx. + mainConsumer, err := NewConsumer(conn, DefaultConsumerConfig(). + WithQueueConfig(DefaultQueueConfig(mainQ).WithDurable(false).WithAutoDelete(true).WithExclusive(true).WithDeadLetter(dlx, ""))) + if err != nil { + t.Fatalf("failed to create main consumer: %v", err) + } + t.Cleanup(func() { mainConsumer.Close() }) + + var attempts atomic.Int32 + go func() { + _ = mainConsumer.Consume(ctx, func(_ context.Context, _ *Delivery) error { + attempts.Add(1) + return errors.New("always fails") + }) + }() + + if err := pub.Publish(ctx, NewTextMessage("poison")); err != nil { + t.Fatalf("publish: %v", err) + } + + // The rejected message must be dead-lettered to the DLQ. + recvDelivery(t, ctx, dlqCh, "poison") + + // And the handler must not have hot-looped: give it a moment, then assert + // it ran exactly once (a requeue bug would show many attempts). + time.Sleep(300 * time.Millisecond) + if n := attempts.Load(); n != 1 { + t.Errorf("handler ran %d times, want exactly 1 (no requeue hot-loop)", n) + } +} + +// TestIntegration_ErrRequeueRedelivers verifies that a handler returning +// ErrRequeue causes the message to be requeued even under the default +// RequeueOnError=false, and that it is redelivered and eventually acked. +func TestIntegration_ErrRequeueRedelivers(t *testing.T) { + conn := integrationConn(t) + queue := uniqueQueue(t) + + consumer, err := NewConsumer(conn, DefaultConsumerConfig(). + WithQueueConfig(DefaultQueueConfig(queue).WithDurable(false).WithAutoDelete(true).WithExclusive(true))) + if err != nil { + t.Fatalf("failed to create consumer: %v", err) + } + t.Cleanup(func() { consumer.Close() }) + + pub, err := NewPublisher(conn, DefaultPublisherConfig().WithRoutingKey(queue)) + if err != nil { + t.Fatalf("failed to create publisher: %v", err) + } + t.Cleanup(func() { pub.Close() }) + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + var attempts atomic.Int32 + done := make(chan struct{}) + go func() { + _ = consumer.Consume(ctx, func(_ context.Context, _ *Delivery) error { + switch attempts.Add(1) { + case 1: + // Transient failure: force a requeue despite the false default. + return fmt.Errorf("transient: %w", ErrRequeue) + case 2: + close(done) + } + return nil // success on redelivery → ack + }) + }() + + if err := pub.Publish(ctx, NewTextMessage("retry-me")); err != nil { + t.Fatalf("publish: %v", err) + } + + select { + case <-done: + case <-ctx.Done(): + t.Fatalf("timed out; handler ran %d time(s), expected a redelivery", attempts.Load()) + } + if n := attempts.Load(); n < 2 { + t.Errorf("handler ran %d times, want >= 2 (requeued then acked)", n) + } +} diff --git a/middleware.go b/middleware.go index d4e196a..803c7b0 100644 --- a/middleware.go +++ b/middleware.go @@ -67,6 +67,14 @@ func LoggingMiddleware(logger Logger) Middleware { // RetryMiddleware retries failed message processing up to maxRetries times // with the given delay between attempts. A negative maxRetries is treated as // zero, so the handler always runs at least once. +// +// Retries happen in-process: the handler goroutine (and its prefetch slot) is +// held for the whole delay, so this suits short retries, not long backoff. +// After the retries are exhausted the final error propagates to the consumer's +// normal disposition (see ConsumerConfig.RequeueOnError): with the default it +// is rejected and dead-lettered. Pairing this with RequeueOnError=true, and not +// returning ErrDrop from the handler, causes the message to be requeued and +// retried again — an unbounded loop. func RetryMiddleware(maxRetries int, delay time.Duration) Middleware { if maxRetries < 0 { maxRetries = 0 diff --git a/middleware_test.go b/middleware_test.go index 4cdecaf..a8f44fd 100644 --- a/middleware_test.go +++ b/middleware_test.go @@ -3,6 +3,7 @@ package rabbitmq import ( "context" "errors" + "fmt" "strings" "sync/atomic" "testing" @@ -144,6 +145,30 @@ func TestRetryMiddlewareExhausted(t *testing.T) { } } +// TestRetryMiddlewarePreservesDisposition ensures a disposition sentinel +// returned by the handler survives being propagated through RetryMiddleware, +// so the consumer's requeue decision still sees it after retries are exhausted. +func TestRetryMiddlewarePreservesDisposition(t *testing.T) { + dropHandler := RetryMiddleware(2, 1*time.Millisecond)(func(_ context.Context, _ *Delivery) error { + return fmt.Errorf("poison: %w", ErrDrop) + }) + err := dropHandler(context.Background(), &Delivery{Message: &Message{}}) + if !errors.Is(err, ErrDrop) { + t.Errorf("expected ErrDrop to survive RetryMiddleware, got %v", err) + } + if requeueDecision(err, true) { + t.Error("ErrDrop after retries should force no-requeue even with default true") + } + + requeueHandler := RetryMiddleware(2, 1*time.Millisecond)(func(_ context.Context, _ *Delivery) error { + return fmt.Errorf("transient: %w", ErrRequeue) + }) + err = requeueHandler(context.Background(), &Delivery{Message: &Message{}}) + if !requeueDecision(err, false) { + t.Error("ErrRequeue after retries should force requeue even with default false") + } +} + func TestRetryMiddlewareContextCancelled(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() diff --git a/rabbitmq.go b/rabbitmq.go index bc7e332..41744aa 100644 --- a/rabbitmq.go +++ b/rabbitmq.go @@ -35,6 +35,15 @@ var ( // ErrDelayTooLong is returned by PublishDelayed when the requested delay // exceeds the largest rung of the delay ladder (see DelayLadder). ErrDelayTooLong = errors.New("rabbitmq: delay exceeds maximum supported delay") + // ErrRequeue, when returned by a message handler, forces the failed message + // to be requeued regardless of ConsumerConfig.RequeueOnError. Use it for + // transient failures that are worth retrying. May be wrapped with %w. + ErrRequeue = errors.New("rabbitmq: requeue message") + // ErrDrop, when returned by a message handler, forces the failed message to + // NOT be requeued regardless of ConsumerConfig.RequeueOnError. The message + // is dead-lettered if a dead-letter exchange is configured, else discarded. + // Use it for poison messages that will never succeed. May be wrapped with %w. + ErrDrop = errors.New("rabbitmq: drop message") ) // Config holds the RabbitMQ connection configuration. From 7917c899f1bbbcfb7f35e219dc5f08c19921a643 Mon Sep 17 00:00:00 2001 From: kartik Date: Sun, 5 Jul 2026 09:36:12 +0530 Subject: [PATCH 2/2] test/docs: address review feedback on requeue-disposition change MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Cover the previously-untested nack-failure branch in processDelivery with a focused unit test (a Delivery with a nil Acknowledger forces Nack to error), asserting OnError receives the nack error. - Replace the hot-loop check's single time.Sleep with bounded polling that fails the instant the attempt count climbs past one — deterministic, not timing-luck. - Correct RetryMiddleware godoc: after retries exhaust, the message is dead-lettered only if the queue has a DLX (else discarded), and note that every error is retried, with ErrRequeue/ErrDrop affecting only the final disposition. --- consumer_test.go | 32 ++++++++++++++++++++++++++++++++ integration_test.go | 14 +++++++++----- middleware.go | 11 ++++++++--- 3 files changed, 49 insertions(+), 8 deletions(-) diff --git a/consumer_test.go b/consumer_test.go index 83d3732..3c083d9 100644 --- a/consumer_test.go +++ b/consumer_test.go @@ -1,6 +1,7 @@ package rabbitmq import ( + "context" "errors" "fmt" "testing" @@ -41,6 +42,37 @@ func TestDefaultConsumerConfig(t *testing.T) { } } +// TestProcessDeliveryNackFailureReported ensures that when Nack fails (here +// forced by a Delivery with no Acknowledger), the nack error is surfaced to the +// configured OnError handler rather than swallowed. +func TestProcessDeliveryNackFailureReported(t *testing.T) { + var reported []error + c := &Consumer{config: ConsumerConfig{ + AutoAck: false, + RequeueOnError: false, + OnError: func(err error) { reported = append(reported, err) }, + }} + + handlerErr := errors.New("handler failed") + // Zero-value embedded amqp.Delivery has a nil Acknowledger, so Nack errors. + d := &Delivery{Message: &Message{}} + + c.processDelivery(context.Background(), func(_ context.Context, _ *Delivery) error { + return handlerErr + }, d) + + // OnError is called once for the handler error and once for the nack error. + if len(reported) != 2 { + t.Fatalf("expected 2 OnError calls (handler + nack), got %d: %v", len(reported), reported) + } + if !errors.Is(reported[0], handlerErr) { + t.Errorf("first OnError = %v, want handler error", reported[0]) + } + if reported[1] == nil || errors.Is(reported[1], handlerErr) { + t.Errorf("second OnError = %v, want a non-nil nack error", reported[1]) + } +} + func TestRequeueDecision(t *testing.T) { tests := []struct { name string diff --git a/integration_test.go b/integration_test.go index 379b41a..5160cab 100644 --- a/integration_test.go +++ b/integration_test.go @@ -2122,11 +2122,15 @@ func TestIntegration_TerminalErrorNotRequeued(t *testing.T) { // The rejected message must be dead-lettered to the DLQ. recvDelivery(t, ctx, dlqCh, "poison") - // And the handler must not have hot-looped: give it a moment, then assert - // it ran exactly once (a requeue bug would show many attempts). - time.Sleep(300 * time.Millisecond) - if n := attempts.Load(); n != 1 { - t.Errorf("handler ran %d times, want exactly 1 (no requeue hot-loop)", n) + // And the handler must not have hot-looped: the handler already ran once + // (the message reached the DLQ), so sample the count over a short window and + // fail the instant it climbs past one. A requeue bug would keep incrementing. + deadline := time.Now().Add(500 * time.Millisecond) + for time.Now().Before(deadline) { + if n := attempts.Load(); n != 1 { + t.Fatalf("handler ran %d times, want exactly 1 (no requeue hot-loop)", n) + } + time.Sleep(50 * time.Millisecond) } } diff --git a/middleware.go b/middleware.go index 803c7b0..952b301 100644 --- a/middleware.go +++ b/middleware.go @@ -70,11 +70,16 @@ func LoggingMiddleware(logger Logger) Middleware { // // Retries happen in-process: the handler goroutine (and its prefetch slot) is // held for the whole delay, so this suits short retries, not long backoff. +// Every non-nil error is retried — including ErrRequeue and ErrDrop; those +// sentinels only affect the final disposition once the retries are exhausted, +// not whether a retry happens. +// // After the retries are exhausted the final error propagates to the consumer's // normal disposition (see ConsumerConfig.RequeueOnError): with the default it -// is rejected and dead-lettered. Pairing this with RequeueOnError=true, and not -// returning ErrDrop from the handler, causes the message to be requeued and -// retried again — an unbounded loop. +// is rejected, and dead-lettered if the queue has a dead-letter exchange, else +// discarded. Pairing this with RequeueOnError=true, and not returning ErrDrop +// from the handler, causes the message to be requeued and retried again — an +// unbounded loop. func RetryMiddleware(maxRetries int, delay time.Duration) Middleware { if maxRetries < 0 { maxRetries = 0