feat(consumer): safe requeue default + per-message disposition sentinels

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

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

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,14 +491,28 @@ 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 {
 		if c.config.OnError != nil {
 			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)
 				}

consumer_test.go

@@ -1,7 +1,9 @@
 package rabbitmq
 
 import (
+	"context"
 	"errors"
+	"fmt"
 	"testing"
 	"time"
 )
@@ -35,8 +37,64 @@ 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)")
+	}
+}
+
+// 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
+		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)
+			}
+		})
 	}
 }
 

integration_test.go

@@ -5,6 +5,7 @@ package rabbitmq
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"os"
 	"sync"
@@ -2057,3 +2058,130 @@ 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: 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)
+	}
+}
+
+// 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)
+	}
+}

middleware.go

@@ -67,6 +67,19 @@ 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.
+// 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 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