chore: added tests to expose pending list bug in rediq

Author: subomi

consumer_test.go

@@ -1,10 +1,14 @@
 package taskq_test
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
+	"io"
+	"log"
 	"math/rand"
+	"os"
 	"runtime"
 	"strconv"
 	"strings"
@@ -92,6 +96,77 @@ func testConsumer(t *testing.T, factory taskq.Factory, opt *taskq.QueueOptions)
 	}
 }
 
+func testConsumerDelete(t *testing.T, factory taskq.Factory, opt *taskq.QueueOptions) {
+	old := os.Stderr
+	r, w, err := os.Pipe()
+	if err != nil {
+		log.Fatal(err)
+	}
+	os.Stderr = w
+
+	taskq.SetLogger(log.New(os.Stderr, "taskq: ", log.LstdFlags|log.Lshortfile))
+
+	c := context.Background()
+	opt.WaitTimeout = waitTimeout
+	opt.Redis = redisRing()
+
+	q := factory.RegisterQueue(opt)
+	defer q.Close()
+
+	purge(t, q)
+
+	outC := make(chan string)
+	go func() {
+		var buf bytes.Buffer
+		io.Copy(&buf, r)
+		outC <- buf.String()
+	}()
+
+	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")
+	}
+
+	time.Sleep(2 * time.Second)
+	w.Close()
+	os.Stderr = old
+
+	out := <-outC
+	findPendingLogLine := strings.Contains(out, "redisq: pending failed: redisq: can't find pending message")
+
+	if findPendingLogLine {
+		t.Fatalf("data 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

redisq/queue.go

@@ -377,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
@@ -390,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"),