Merge pull request #145 from Subomi/fix/ack-processed-messages-from-redisq

Call XAck inside Delete in redsiq

Author: vmihailenco

consumer_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
 	"math/rand"
 	"runtime"
 	"strconv"
@@ -17,6 +18,7 @@ import (
 	"github.com/go-redis/redis_rate/v9"
 
 	"github.com/vmihailenco/taskq/v3"
+	"github.com/vmihailenco/taskq/v3/redisq"
 )
 
 const waitTimeout = time.Second
@@ -92,6 +94,73 @@ func testConsumer(t *testing.T, factory taskq.Factory, opt *taskq.QueueOptions)
 	}
 }
 
+func testConsumerDelete(t *testing.T, factory taskq.Factory, opt *taskq.QueueOptions) {
+	c := context.Background()
+	opt.WaitTimeout = waitTimeout
+	opt.Redis = redisRing()
+
+	red, ok := opt.Redis.(redisq.RedisStreamClient)
+	if !ok {
+		log.Fatal(fmt.Errorf("redisq: Redis client must support streams"))
+	}
+
+	q := factory.RegisterQueue(opt)
+	defer q.Close()
+
+	purge(t, q)
+
+	ch := make(chan time.Time)
+	task := taskq.RegisterTask(&taskq.TaskOptions{
+		Name: nextTaskID(),
+		Handler: func() error {
+			ch <- time.Now()
+			return nil
+		},
+	})
+
+	err := q.Add(task.WithArgs(c))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	p := q.Consumer()
+	if err := p.Start(c); err != nil {
+		t.Fatal(err)
+	}
+
+	select {
+	case <-ch:
+	case <-time.After(testTimeout):
+		t.Fatalf("message was not processed")
+	}
+
+	tm := time.Now().Add(opt.ReservationTimeout)
+	end := strconv.FormatInt(unixMs(tm), 10)
+	pending, err := red.XPendingExt(context.Background(), &redis.XPendingExtArgs{
+		Stream: "taskq:{" + opt.Name + "}:stream",
+		Group:  "taskq",
+		Start:  "-",
+		End:    end,
+		Count:  100,
+	}).Result()
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(pending) > 0 {
+		t.Fatal("task not acknowledged and still exists in pending list.")
+	}
+
+	if err := p.Stop(); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := q.Close(); err != nil {
+		t.Fatal(err)
+	}
+}
+
 func testUnknownTask(t *testing.T, factory taskq.Factory, opt *taskq.QueueOptions) {
 	c := context.Background()
 	opt.WaitTimeout = waitTimeout
@@ -763,3 +832,7 @@ func nextTaskID() string {
 	taskID++
 	return id
 }
+
+func unixMs(tm time.Time) int64 {
+	return tm.UnixNano() / int64(time.Millisecond)
+}

redisq/queue.go

@@ -23,7 +23,7 @@ import (
 
 const batchSize = 100
 
-type redisStreamClient interface {
+type RedisStreamClient interface {
 	Del(ctx context.Context, keys ...string) *redis.IntCmd
 	TxPipeline() redis.Pipeliner
 
@@ -49,7 +49,7 @@ type Queue struct {
 
 	consumer *taskq.Consumer
 
-	redis redisStreamClient
+	redis RedisStreamClient
 	wg    sync.WaitGroup
 
 	zset                string
@@ -73,7 +73,7 @@ func NewQueue(opt *taskq.QueueOptions) *Queue {
 	if opt.Redis == nil {
 		panic(fmt.Errorf("redisq: Redis client is required"))
 	}
-	red, ok := opt.Redis.(redisStreamClient)
+	red, ok := opt.Redis.(RedisStreamClient)
 	if !ok {
 		panic(fmt.Errorf("redisq: Redis client must support streams"))
 	}
@@ -147,7 +147,7 @@ func (q *Queue) Add(msg *taskq.Message) error {
 	return q.add(q.redis, msg)
 }
 
-func (q *Queue) add(pipe redisStreamClient, msg *taskq.Message) error {
+func (q *Queue) add(pipe RedisStreamClient, msg *taskq.Message) error {
 	if msg.TaskName == "" {
 		return internal.ErrTaskNameRequired
 	}
@@ -243,6 +243,10 @@ func (q *Queue) Release(msg *taskq.Message) error {
 
 // Delete deletes the message from the queue.
 func (q *Queue) Delete(msg *taskq.Message) error {
+	err := q.redis.XAck(msg.Ctx, q.stream, q.streamGroup, msg.ID).Err()
+	if err != nil {
+		return err
+	}
 	return q.redis.XDel(msg.Ctx, q.stream, msg.ID).Err()
 }
 
@@ -373,6 +377,7 @@ func (q *Queue) schedulePending(ctx context.Context) (int, error) {
 	if err != nil {
 		if strings.HasPrefix(err.Error(), "NOGROUP") {
 			q.createStreamGroup(ctx)
+
 			return 0, nil
 		}
 		return 0, err
@@ -386,8 +391,9 @@ func (q *Queue) schedulePending(ctx context.Context) (int, error) {
 		if err != nil {
 			return 0, err
 		}
+
 		if len(xmsgs) != 1 {
-			err := fmt.Errorf("redisq: can't find peding message id=%q in stream=%q",
+			err := fmt.Errorf("redisq: can't find pending message id=%q in stream=%q",
 				id, q.stream)
 			return 0, err
 		}

redisq_test.go

@@ -2,6 +2,7 @@ package taskq_test
 
 import (
 	"testing"
+	"time"
 
 	"github.com/vmihailenco/taskq/v3"
 	"github.com/vmihailenco/taskq/v3/redisq"
@@ -17,6 +18,13 @@ func TestRedisqConsumer(t *testing.T) {
 	})
 }
 
+func TestRedisqAckMessage(t *testing.T) {
+	testConsumerDelete(t, redisqFactory(), &taskq.QueueOptions{
+		Name:               queueName("redisq-ack-message"),
+		ReservationTimeout: 1 * time.Second,
+	})
+}
+
 func TestRedisqUnknownTask(t *testing.T) {
 	testUnknownTask(t, redisqFactory(), &taskq.QueueOptions{
 		Name: queueName("redisq-unknown-task"),