go-omni/queue/redis/consumer_test.go

package redis

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"git.yuntuo.hk/k2safe/go-omni/queue"
	goredis "github.com/redis/go-redis/v9"
	"os"
	"sync/atomic"
	"testing"
	"time"
)

func testStore(t *testing.T, max int) *Store {
	t.Helper()
	address := os.Getenv("GO_OMNI_REDIS_TEST_ADDR")
	if address == "" {
		t.Skip("requires isolated Redis")
	}
	client := goredis.NewClient(&goredis.Options{Addr: address, PoolSize: 4, MaxRetries: -1, ContextTimeoutEnabled: true, ReadTimeout: time.Second, WriteTimeout: time.Second})
	s, err := New(client, Config{Namespace: fmt.Sprint(t.Name(), time.Now().UnixNano()), MaxRecords: max, DedupTTL: time.Second})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { client.Del(context.Background(), s.keys...); client.Close() })
	return s
}
func message(id string) queue.Message {
	return queue.Message{ID: id, Name: "test", Payload: json.RawMessage(`{"value":1}`)}
}
func eventually(t *testing.T, f func() bool) {
	t.Helper()
	deadline := time.Now().Add(3 * time.Second)
	for !f() {
		if time.Now().After(deadline) {
			t.Fatal("condition did not recover")
		}
		time.Sleep(10 * time.Millisecond)
	}
}
func TestQueueRedisIntegration(t *testing.T) {
	t.Run("publicationAndCapacity", func(t *testing.T) {
		s := testStore(t, 1)
		ctx := context.Background()
		receipt, err := s.Publish(ctx, message("first"))
		if err != nil {
			t.Fatal(err)
		}
		again, err := s.Publish(ctx, message("first"))
		if err != nil || again != receipt {
			t.Fatal(again, err)
		}
		changed := message("first")
		changed.Payload = json.RawMessage(`2`)
		if _, err = s.Publish(ctx, changed); !errors.Is(err, queue.ErrConflict) {
			t.Fatal(err)
		}
		if _, err = s.Publish(ctx, message("second")); !errors.Is(err, queue.ErrFull) {
			t.Fatal(err)
		}
	})
	t.Run("staleAcknowledgement", func(t *testing.T) {
		s := testStore(t, 10)
		ctx := context.Background()
		receipt, _ := s.Publish(ctx, message("first"))
		s.client.XGroupCreateMkStream(ctx, s.keys[0], "workers", "0")
		s.client.XReadGroup(ctx, &goredis.XReadGroupArgs{Group: "workers", Consumer: "old", Streams: []string{s.keys[0], ">"}, Count: 1, Block: -1})
		if err := s.client.XClaim(ctx, &goredis.XClaimArgs{Stream: s.keys[0], Group: "workers", Consumer: "new", MinIdle: 0, Messages: []string{receipt.BackendID}}).Err(); err != nil {
			t.Fatal(err)
		}
		delivery := queue.Delivery{Message: message("first"), Attempt: 2}
		if err := s.finish(ctx, receipt.BackendID, "old", delivery, ""); !errors.Is(err, queue.ErrLeaseLost) {
			t.Fatal(err)
		}
		if err := s.finish(ctx, receipt.BackendID, "new", delivery, ""); err != nil {
			t.Fatal(err)
		}
		if n := s.client.XLen(ctx, s.keys[0]).Val(); n != 0 {
			t.Fatal(n)
		}
	})
	t.Run("crashReclaimRetryAndDeadLetters", func(t *testing.T) {
		s := testStore(t, 10)
		ctx := context.Background()
		s.Publish(ctx, message("crashed"))
		s.Publish(ctx, message("retry"))
		s.Publish(ctx, message("reject"))
		s.client.XGroupCreateMkStream(ctx, s.keys[0], "workers", "0")
		s.client.XReadGroup(ctx, &goredis.XReadGroupArgs{Group: "workers", Consumer: "crashedProcess", Streams: []string{s.keys[0], ">"}, Count: 1, Block: -1})
		cfg := ConsumerConfig{Workers: 1, MaxAttempts: 3, TaskTimeout: 50 * time.Millisecond, VisibilityTimeout: 100 * time.Millisecond, ShutdownTimeout: time.Second}
		consumer, _ := NewConsumer(s, cfg)
		second, _ := NewConsumer(s, cfg)
		runCtx, cancel := context.WithCancel(ctx)
		var retries, reclaimed, rejected atomic.Int32
		handler := func(ctx context.Context, d queue.Delivery) error {
			switch d.ID {
			case "crashed":
				if d.Attempt != 2 {
					t.Error("reclaim count", d.Attempt)
				}
				reclaimed.Add(1)
				return nil
			case "retry":
				retries.Add(1)
				return errors.New("known retryable failure")
			default:
				rejected.Add(1)
				return queue.Reject("invalidInput")
			}
		}
		done := make(chan error, 2)
		go func() { done <- consumer.Run(runCtx, handler) }()
		go func() { done <- second.Run(runCtx, handler) }()
		defer cancel()
		eventually(t, func() bool {
			dead, err := s.DeadLetters(ctx, "", 100)
			return err == nil && len(dead) == 2 && reclaimed.Load() == 1
		})
		cancel()
		for i := 0; i < 2; i++ {
			if err := <-done; !errors.Is(err, context.Canceled) {
				t.Fatal(err)
			}
		}
		if retries.Load() != 3 || rejected.Load() != 1 {
			t.Fatal(retries.Load(), rejected.Load())
		}
		dead, err := s.DeadLetters(ctx, "", 1)
		if err != nil || len(dead) != 1 {
			t.Fatal(dead, err)
		}
		rest, err := s.DeadLetters(ctx, dead[0].StreamID, 1)
		if err != nil || len(rest) != 1 || rest[0].StreamID == dead[0].StreamID {
			t.Fatal(rest, err)
		}
		if err = s.DiscardDead(ctx, dead[0].Message.ID, dead[0].StreamID); err != nil {
			t.Fatal(err)
		}
		remaining, _ := s.DeadLetters(ctx, "", 100)
		if len(remaining) != 1 {
			t.Fatal(remaining)
		}
		// Retained completed IDs expire and release publication capacity.
		eventually(t, func() bool {
			_, err := s.Publish(ctx, message("crashed"))
			if err != nil {
				return false
			}
			return s.client.XLen(ctx, s.keys[0]).Val() == 2
		})
	})
}