- max delivery count

- minid, limit parameters support in the produces level
- minid, limit parameters support in the enqueue level

Author: raimdev

consumer.go

@@ -2,6 +2,7 @@ package redisqueue
 
 import (
 	"context"
+	"log/slog"
 	"net"
 	"os"
 	"sync"
@@ -53,6 +54,12 @@ type ConsumerOptions struct {
 	BufferSize int
 	// Concurrency dictates how many goroutines to spawn to handle the messages.
 	Concurrency int
+
+	// MaxDeliveryCount is the maximum number of times a message can be delivered
+	// before it is considered failed. If this is set to 0, the message will be
+	// retried indefinitely
+	MaxDeliveryCount int64
+
 	// RedisClient supersedes the RedisOptions field, and allows you to inject
 	// an already-made Redis Client for use in the consumer. This may be either
 	// the standard client or a cluster client.
@@ -276,6 +283,15 @@ func (c *Consumer) reclaim(ctx context.Context) {
 					msgs := make([]string, 0)
 
 					for _, r := range res {
+						slog.Info("pending message", "id", r.ID, "count", r.RetryCount, "max", c.options.MaxDeliveryCount)
+						if c.options.MaxDeliveryCount > 0 && r.RetryCount >= c.options.MaxDeliveryCount {
+							slog.Info("message exceeded delivery count limit", "id", r.ID, "count", r.RetryCount, "max", c.options.MaxDeliveryCount)
+							err = c.redis.XAck(ctx, stream, c.options.GroupName, r.ID).Err()
+							if err != nil {
+								c.Errors <- errors.Wrapf(err, "error acknowledging after retry count exceeded for %q stream and %q message, ", stream, r.ID)
+								continue
+							}
+						}
 						if r.Idle >= c.options.VisibilityTimeout {
 							claimres, err := c.redis.XClaim(ctx, &redis.XClaimArgs{
 								Stream:   stream,
@@ -331,7 +347,7 @@ func (c *Consumer) poll(ctx context.Context) {
 		case <-c.stopPoll:
 			// once the polling has stopped (i.e. there will be no more messages
 			// put onto c.queue), stop all of the workers
-			for i := 0; i < c.options.Concurrency; i++ {
+			for range c.options.Concurrency {
 				c.stopWorkers <- struct{}{}
 			}
 			return

consumer_test.go

@@ -2,6 +2,8 @@ package redisqueue
 
 import (
 	"context"
+	"fmt"
+	"log/slog"
 	"os"
 	"testing"
 	"time"
@@ -249,6 +251,67 @@ func TestRun(t *testing.T) {
 		c.Run(ctx)
 	})
 
+	t.Run("reclaims pending messages maximum MaxDeliveryCount", func(tt *testing.T) {
+		ctx := context.Background()
+		var maxDeliveryCount int64 = 3
+		var visibilityTimeout time.Duration = 5 * time.Millisecond
+		var reclaimInterval time.Duration = 1 * time.Millisecond
+
+		// create a consumer
+		c, err := NewConsumerWithOptions(ctx, &ConsumerOptions{
+			VisibilityTimeout: visibilityTimeout,
+			BlockingTimeout:   10 * time.Millisecond,
+			ReclaimInterval:   1 * time.Millisecond,
+			BufferSize:        100,
+			Concurrency:       10,
+			MaxDeliveryCount:  maxDeliveryCount,
+		})
+		require.NoError(tt, err)
+
+		// create a producer
+		p, err := NewProducer()
+		require.NoError(tt, err)
+
+		// create consumer group
+		c.redis.XGroupDestroy(ctx, tt.Name(), c.options.GroupName)
+		c.redis.XGroupCreateMkStream(ctx, tt.Name(), c.options.GroupName, "$")
+
+		// enqueue a message
+		msg := &Message{
+			Stream: tt.Name(),
+			Values: map[string]interface{}{"test": "value"},
+		}
+		err = p.Enqueue(ctx, msg)
+		require.NoError(tt, err)
+
+		var deliveryCount int64 = 0
+
+		// register a handler that will assert the message and then shut down
+		// the consumer
+		c.Register(tt.Name(), func(m *Message) error {
+			slog.Info("message received", "id", m.ID, "delivery count", deliveryCount)
+
+			deliveryCount++
+			assert.Equal(tt, msg.ID, m.ID)
+			return fmt.Errorf("dummy error")
+		})
+
+		// // watch for consumer errors
+		// go func() {
+		// 	<-c.Errors
+		// }()
+
+		// run the consumer
+		go c.Run(ctx)
+
+		// wait for more than VisibilityTimeout + (ReclaimInterval*number_higher_than_max_dlivery_count) to ensure that
+		// the message was reclaimed more than MaxDeliveryCount
+
+		time.Sleep(visibilityTimeout + (reclaimInterval * 10) + 1*time.Millisecond)
+		c.Shutdown()
+		assert.Equal(tt, maxDeliveryCount, deliveryCount)
+	})
+
 	t.Run("doesn't reclaim if there is no VisibilityTimeout set", func(tt *testing.T) {
 		ctx := context.Background()
 
@@ -263,8 +326,8 @@ func TestRun(t *testing.T) {
 
 		// create a producer
 		p, err := NewProducerWithOptions(ctx, &ProducerOptions{
-			StreamMaxLength:      2,
-			ApproximateMaxLength: false,
+			StreamMaxLength: 2,
+			UseApproximate:  false,
 		})
 		require.NoError(tt, err)
 
@@ -345,8 +408,8 @@ func TestRun(t *testing.T) {
 
 		// create a producer
 		p, err := NewProducerWithOptions(ctx, &ProducerOptions{
-			StreamMaxLength:      1,
-			ApproximateMaxLength: false,
+			StreamMaxLength: 1,
+			UseApproximate:  false,
 		})
 		require.NoError(tt, err)
 

message.go

@@ -4,7 +4,10 @@ package redisqueue
 // When enqueuing, it's recommended to leave ID empty and let Redis generate it,
 // unless you know what you're doing.
 type Message struct {
-	ID     string
-	Stream string
-	Values map[string]interface{}
+	ID              string
+	Stream          string
+	StreamMaxLength int64
+	StreamMinId     string
+	TrimLimit       int64
+	Values          map[string]interface{}
 }

mise.toml

@@ -0,0 +1,2 @@
+[tools]
+go = "1.23"

producer.go

@@ -1,6 +1,7 @@
 package redisqueue
 
 import (
+	"cmp"
 	"context"
 
 	"github.com/redis/go-redis/v9"
@@ -17,10 +18,20 @@ type ProducerOptions struct {
 	// So ideally, you'll set this number to be as high as you can makee it.
 	// More info here: https://redis.io/commands/xadd#capped-streams.
 	StreamMaxLength int64
-	// ApproximateMaxLength determines whether to use the ~ with the MAXLEN
+
+	// StreamMinId sets the minimum ID that will be used when calling XADD. This
+	// is useful when you want to ensure that the stream is trimmed to a certain
+	// point. More info here: https://redis.io/commands/xadd#capped-streams.
+	StreamMinId string
+
+	// UseApproximate determines whether to use the ~ with the MAXLEN and MINID
 	// option. This allows the stream trimming to done in a more efficient
 	// manner. More info here: https://redis.io/commands/xadd#capped-streams.
-	ApproximateMaxLength bool
+	UseApproximate bool
+
+	// TrimLimit sets LIMIT for XADD
+	TrimLimit int64
+
 	// RedisClient supersedes the RedisOptions field, and allows you to inject
 	// an already-made Redis Client for use in the consumer. This may be either
 	// the standard client or a cluster client.
@@ -41,12 +52,12 @@ type Producer struct {
 }
 
 var defaultProducerOptions = &ProducerOptions{
-	StreamMaxLength:      1000,
-	ApproximateMaxLength: true,
+	StreamMaxLength: 1000,
+	UseApproximate:  true,
 }
 
 // NewProducer uses a default set of options to create a Producer. It sets
-// StreamMaxLength to 1000 and ApproximateMaxLength to true. In most production
+// StreamMaxLength to 1000 and UseApproximate to true. In most production
 // environments, you'll want to use NewProducerWithOptions.
 func NewProducer() (*Producer, error) {
 	return NewProducerWithOptions(context.Background(), defaultProducerOptions)
@@ -77,12 +88,20 @@ func NewProducerWithOptions(ctx context.Context, options *ProducerOptions) (*Pro
 // should let Redis auto-generate the ID. If an ID is auto-generated, it will be
 // set on msg.ID for your reference. msg.Values is also required.
 func (p *Producer) Enqueue(ctx context.Context, msg *Message) error {
+	maxLen := cmp.Or(msg.StreamMaxLength, p.options.StreamMaxLength)
+	minId := cmp.Or(msg.StreamMinId, p.options.StreamMinId)
+	if maxLen > 0 {
+		minId = ""
+	}
+
 	args := &redis.XAddArgs{
 		ID:     msg.ID,
 		Stream: msg.Stream,
 		Values: msg.Values,
-		MaxLen: p.options.StreamMaxLength,
-		Approx: p.options.ApproximateMaxLength,
+		MaxLen: maxLen,
+		MinID:  minId,
+		Approx: p.options.UseApproximate,
+		Limit:  cmp.Or(msg.TrimLimit, p.options.TrimLimit),
 	}
 	id, err := p.redis.XAdd(ctx, args).Result()
 	if err != nil {

producer_test.go

@@ -67,7 +67,7 @@ func TestEnqueue(t *testing.T) {
 
 	t.Run("bubbles up errors", func(tt *testing.T) {
 		ctx := context.Background()
-		p, err := NewProducerWithOptions(context.Background(), &ProducerOptions{ApproximateMaxLength: true})
+		p, err := NewProducerWithOptions(context.Background(), &ProducerOptions{UseApproximate: true})
 		require.NoError(t, err)
 
 		msg := &Message{