XClaim does not work as expected

[piavgh]

Issue tracker is used for reporting bugs and discussing new features. Please use
stackoverflow for supporting issues.

EDIT: this is not a bug but an expected behavior of Redis Stream
Related issues:

• XCLAIMing XDELeted entries redis#7021
• [BUG] XAUTOCLAIM returns nil for old messages in PEL redis#10198
• [BUG] xclaim claims the message but returns nil redis#8924

---

Expected Behavior

The XClaim method should return the list of claimed messages (XMessages) (as specified in the document https://redis.io/commands/xclaim/ Return section), and err = nil in case of success event

Current Behavior

• It claim the messages successfully (I checked RedisInsight and I see that the messages were claimed)
  

• The returned XMessages is nil, while the err = redis: nil
  

Possible Solution

Steps to Reproduce

xMessages, err := s.client.XClaim(ctx, &redis.XClaimArgs{
		Stream:  "stream-name",
		Group:    "consumer-group-name",
		Consumer: "consumer-name",
		MinIdle:  30 * time.Second,
		Messages: <list of messages>,
	}).Result()
	if err != nil {
		return nil, err // <= it will run into this section
	}
 
        ...

Context (Environment)

I'm trying to use XPendingExt and XClaim to set up a workflow that if one consumer dies unexpectedly, another consumer can take the unprocessed messages and continue to process them (so that no messages are loss without being processed)

Detailed Description

Possible Implementation

[monkey92t]

my env

go-redis: v9.0.2
redis-version: 7.0.8 single node
go-version: 1.20

It works well in the example below.

example:

	ctx := context.Background()
	client := redis.NewClient(&redis.Options{
		Addr: ":6379",
	})

	_, err := client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
		pipe.Del(ctx, "stream-name")

		pipe.XAdd(ctx, &redis.XAddArgs{
			Stream: "stream-name",
			ID:     "1-0",
			Values: map[string]interface{}{"a-1": "a-2"},
		})
		pipe.XAdd(ctx, &redis.XAddArgs{
			Stream: "stream-name",
			ID:     "2-0",
			Values: []interface{}{"b-1", "b-2"},
		})
		pipe.XAdd(ctx, &redis.XAddArgs{
			Stream: "stream-name",
			ID:     "3-0",
			Values: []string{"c-1", "c-2"},
		})
		return nil
	})
	if err != nil {
		log.Fatal(err)
	}

	if err = client.XGroupCreate(ctx, "stream-name", "consumer-group-name", "0").Err(); err != nil {
		log.Fatal(err)
	}

	res, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{
		Group:    "consumer-group-name",
		Consumer: "consumer-name",
		Streams:  []string{"stream-name", ">"},
		Count:    2,
	}).Result()
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("x read group:", res)
        // x read group: [{stream-name [{1-0 map[a-1:a-2]} {2-0 map[b-1:b-2]}]}]

	// no ack

	time.Sleep(4 * time.Second)
	
	msgs, err := client.XClaim(ctx, &redis.XClaimArgs{
		Stream:   "stream-name",
		Group:    "consumer-group-name",
		Consumer: "consumer-name",
		MinIdle:  3 * time.Second,
		Messages: []string{"1-0", "2-0", "3-0"},
	}).Result()
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("xClaim messages:", msgs)
        // xClaim messages: [{1-0 map[a-1:a-2]} {2-0 map[b-1:b-2]}]

[monkey92t]

Please note that we paused for 4 seconds in the middle because we set the MinIdle to 3 seconds in our example.

[piavgh]

Please note that we paused for 4 seconds in the middle because we set the MinIdle to 3 seconds in our example.

What do you mean?

[monkey92t]

Please note that we paused for 4 seconds in the middle because we set the MinIdle to 3 seconds in our example.

What do you mean?

redis-doc:

Note that the message is claimed only if its idle time is greater than the minimum idle time we specify when calling XCLAIM. Because as a side effect XCLAIM will also reset the idle time (since this is a new attempt at processing the message), two consumers trying to claim a message at the same time will never both succeed: only one will successfully claim the message. This avoids that we process a given message multiple times in a trivial way (yet multiple processing is possible and unavoidable in the general case).

[monkey92t]

In you example:

&redis.XClaimArgs{
	Stream:  "stream-name",
	Group:    "consumer-group-name",
	Consumer: "consumer-name",
	MinIdle:  30 * time.Second,
	Messages: <list of messages>,
}

MinIdle = 30s

We need to wait for more than 30 seconds for xclaim to get the result.

[piavgh]

@monkey92t : no, you misunderstand my problem.

At the beginning, consumer A (worker-6f7e... in my image) takes 10 messages and stuck => 10 messages are in the pending list of consumer A

After that, consumer B (worker-cda2... in my image`) can claim 10 messages successfully (in my 1st image), but the returned xMessages = nil.
It should return 10 messages in an array

Please reopen this issue

[piavgh]

I will try to use go-redis v9 to see if the issue persists. At the moment, I'm using go-redis v8

[monkey92t]

I still don't understand your question. Could you provide a complete example?

[piavgh]

Let me try to record a video with Goland debugger and RedisInsight. I will try to explain the problem in more detail.

[piavgh]

@monkey92t : please check this video https://screenpal.com/watch/c0fVXfVar45

I'm using go-redis v9.0.2 (latest version) in the video.

Because the code is quite complex and also it's my company asset so it's not easy to create a complete example

[monkey92t]

My example:

STEP-1: I have created three messages and a stream group.

STEP-2: consumer consumer-name-1 read 3 messages. val: 1-0, 2-0, 3-0

STEP-3: sleep 4s, suppose consumer-name-1 is down.

STEP-4: Use the XPending command to check no-xack messages. get: 1-0....3-0

STEP-5: Use the XClaim command to claim message 1-0, new consumer: consumer-name-2

STEP-6: consumer-name-2 successfully processed the message and sent an xack command.

STEP-7: Message 1-0 has been processed, remaining messages 2-0 and 3-0.

func main() {
	ctx := context.Background()
	client := redis.NewClient(&redis.Options{
		Addr: ":6379",
	})

	// STEP-1
	_, err := client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
		pipe.Del(ctx, "stream-name")

		pipe.XAdd(ctx, &redis.XAddArgs{
			Stream: "stream-name",
			ID:     "1-0",
			Values: map[string]interface{}{"a-1": "a-2"},
		})
		pipe.XAdd(ctx, &redis.XAddArgs{
			Stream: "stream-name",
			ID:     "2-0",
			Values: []interface{}{"b-1", "b-2"},
		})
		pipe.XAdd(ctx, &redis.XAddArgs{
			Stream: "stream-name",
			ID:     "3-0",
			Values: []string{"c-1", "c-2"},
		})
		pipe.XGroupCreate(ctx, "stream-name", "consumer-group-name", "0")

		return nil
	})
	if err != nil {
		log.Fatal(err)
	}

	// ---------------------------------------------------------------

	// STEP-2
	res, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{
		Group:    "consumer-group-name",
		Consumer: "consumer-name-1",
		Streams:  []string{"stream-name", ">"},
		Count:    3,
	}).Result()
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("xReadGroup:", res)
	// xReadGroup: [{stream-name [{1-0 map[a-1:a-2]} {2-0 map[b-1:b-2]} {3-0 map[c-1:c-2]}]}]

	// ---------------------------------------------------

	// no ack

	// STEP-3
	time.Sleep(4 * time.Second)

	// ---------------------------------------------------

	// STEP-4
	pending, err := client.XPending(ctx, "stream-name", "consumer-group-name").Result()
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("xPending: %+v\n", pending)
	// xPending: &{Count:3 Lower:1-0 Higher:3-0 Consumers:map[consumer-name-1:3]}

	// ---------------------------------------------------

	// STEP-5
	msgs, err := client.XClaim(ctx, &redis.XClaimArgs{
		Stream:   "stream-name",
		Group:    "consumer-group-name",
		Consumer: "consumer-name-2",
		MinIdle:  3 * time.Second,
		Messages: []string{pending.Lower},
	}).Result()
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("xClaimMessages:", msgs)
	// xClaimMessages: [{1-0 map[a-1:a-2]}]

	// ---------------------------------------------------

	// STEP-6
	err = client.XAck(ctx, "stream-name", "consumer-group-name", pending.Lower).Err()
	if err != nil {
		log.Fatal(err)
	}

	// ---------------------------------------------------

	// STEP-7
	pending, err = client.XPending(ctx, "stream-name", "consumer-group-name").Result()
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("xPending: %+v\n", pending)
	// xPending: &{Count:2 Lower:2-0 Higher:3-0 Consumers:map[consumer-name-1:2]}
}

[monkey92t]

@monkey92t : please check this video https://screenpal.com/watch/c0fVXfVar45

I'm using go-redis v9.0.2 (latest version) in the video.

Because the code is quite complex and also it's my company asset so it's not easy to create a complete example

I watched your video, but I didn't see the code for many functions. I suspect that in your video, the stream message has already been processed. You can use go-redis-hook to print all executed Redis commands.

example:

output:

redis command: [del stream-name]
redis command: [xadd stream-name 1-0 a-1 a-2]
redis command: [xadd stream-name 2-0 b-1 b-2]
redis command: [xadd stream-name 3-0 c-1 c-2]
redis command: [xgroup create stream-name consumer-group-name 0]
redis command: [xreadgroup group consumer-group-name consumer-name-1 count 3 block 0 streams stream-name >]
redis command: [xpending stream-name consumer-group-name]
redis command: [xclaim stream-name consumer-group-name consumer-name-2 3000 0-0]
redis command: [xack stream-name consumer-group-name 1-0]
redis command: [xpending stream-name consumer-group-name]

package main

import (
	"context"
	"fmt"
	"log"
	"net"
	"time"

	"github.com/redis/go-redis/v9"
)

type hook struct{}

func (h *hook) DialHook(next redis.DialHook) redis.DialHook {
	return func(ctx context.Context, network, addr string) (net.Conn, error) {
		return next(ctx, network, addr)
	}
}

func (h *hook) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
	return func(ctx context.Context, cmd redis.Cmder) error {
		fmt.Println("redis command:", cmd.Args())
		return next(ctx, cmd)
	}
}
func (h *hook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
	return func(ctx context.Context, cmds []redis.Cmder) error {
		for _, v := range cmds {
			fmt.Println("redis command:", v.Args())
		}
		return next(ctx, cmds)
	}
}

func main() {
	ctx := context.Background()
	client := redis.NewClient(&redis.Options{
		Addr: ":6379",
	})

	client.AddHook(&hook{})

	// STEP-1
	_, err := client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
		pipe.Del(ctx, "stream-name")

		pipe.XAdd(ctx, &redis.XAddArgs{
			Stream: "stream-name",
			ID:     "1-0",
			Values: map[string]interface{}{"a-1": "a-2"},
		})
		pipe.XAdd(ctx, &redis.XAddArgs{
			Stream: "stream-name",
			ID:     "2-0",
			Values: []interface{}{"b-1", "b-2"},
		})
		pipe.XAdd(ctx, &redis.XAddArgs{
			Stream: "stream-name",
			ID:     "3-0",
			Values: []string{"c-1", "c-2"},
		})
		pipe.XGroupCreate(ctx, "stream-name", "consumer-group-name", "0")

		return nil
	})
	if err != nil {
		log.Fatal(err)
	}

	// ---------------------------------------------------------------

	// STEP-2
	res, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{
		Group:    "consumer-group-name",
		Consumer: "consumer-name-1",
		Streams:  []string{"stream-name", ">"},
		Count:    3,
	}).Result()
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("xReadGroup:", res)
	// xReadGroup: [{stream-name [{1-0 map[a-1:a-2]} {2-0 map[b-1:b-2]} {3-0 map[c-1:c-2]}]}]

	// ---------------------------------------------------

	// no ack

	// STEP-3
	time.Sleep(4 * time.Second)

	// ---------------------------------------------------

	// STEP-4
	pending, err := client.XPending(ctx, "stream-name", "consumer-group-name").Result()
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("xPending: %+v\n", pending)
	// xPending: &{Count:3 Lower:1-0 Higher:3-0 Consumers:map[consumer-name-1:3]}

	// ---------------------------------------------------

	// STEP-5
	msgs, err := client.XClaim(ctx, &redis.XClaimArgs{
		Stream:   "stream-name",
		Group:    "consumer-group-name",
		Consumer: "consumer-name-2",
		MinIdle:  3 * time.Second,
		Messages: []string{pending.Lower},
	}).Result()
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("xClaimMessages:", msgs)
	// xClaimMessages: [{1-0 map[a-1:a-2]}]

	// ---------------------------------------------------

	// STEP-6
	err = client.XAck(ctx, "stream-name", "consumer-group-name", pending.Lower).Err()
	if err != nil {
		log.Fatal(err)
	}

	// ---------------------------------------------------

	// STEP-7
	pending, err = client.XPending(ctx, "stream-name", "consumer-group-name").Result()
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("xPending: %+v\n", pending)
	// xPending: &{Count:2 Lower:2-0 Higher:3-0 Consumers:map[consumer-name-1:2]}
}

[monkey92t]

@piavgh At the end of your video, there were already new consumers taking over the message, so I suspect you repeated the execution of XClaim when handling the messages.

[monkey92t]

I haven't found any further evidence indicating that there's a problem with go-redis.

/cc @vmihailenco What's your opinion?