fix(command): set key position for 16 commands (#4048)

* fix(command): set key position for 16 commands

The shared key resolver routes Ring shards, cluster slots, client-side
caching and full-duplex engines. For typed commands whose key is not
args[1] it fell back to args[1], or, on a warm cluster cache, to the
COMMAND first key, which is 0 for container and numkeys commands:

  BITOP (all BitOp*), MIGRATE           op / host instead of the key
  OBJECT ENCODING/FREQ/IDLETIME/REFCOUNT, XINFO STREAM/GROUPS/CONSUMERS
                                        subcommand instead of the key
  LMPOP, SINTERCARD, ZINTERCARD, ZMPOP  numkeys instead of the key
  BLMPOP, BZMPOP                        timeout instead of the key

Ring sent them to the wrong shard; cluster recovered through MOVED.
The constructors now set the key position, as XGroup*, ZUnion and FCall
already do. A raw FCALL/FCALL_RO also resolves its key after numkeys,
like EVAL, instead of hashing the function name.

* fix(command): route raw forms by their key

The typed constructors of the commands whose key is not args[1] set the
key position; a raw Do / NewCmd of the same commands does not, so the
shared resolver fell back to args[1]: raw XREAD hashed "streams" and raw
LMPOP its numkeys, which picks the wrong Ring shard, the wrong
full-duplex engine, or a cold cluster slot.

The resolver now finds the key for the raw forms: XREAD and XREADGROUP
after STREAMS (XREADGROUP scans past the group and consumer names),
OBJECT and XINFO (HELP is keyless), BITOP, LMPOP, ZMPOP, SINTERCARD,
ZINTERCARD, BLMPOP, BZMPOP, and MIGRATE, including its KEYS form.

* fix(command): walk MIGRATE options for KEYS

The raw MIGRATE KEYS form was located by scanning for the first "keys"
token after the timeout, so an AUTH or AUTH2 operand that reads "keys"
was taken for the clause and the resolver routed by the token after it.
It now walks the option grammar: COPY and REPLACE take no operand,
AUTH takes one and AUTH2 two, and the key follows KEYS.

bitmap_commands.go

@@ -78,6 +78,7 @@ func (c cmdable) bitOp(ctx context.Context, op, destKey string, keys ...string)
 		args[3+i] = key
 	}
 	cmd := NewIntCmd(ctx, args...)
+	cmd.SetFirstKeyPos(2) // the key is not args[1]: route by it
 	_ = c(ctx, cmd)
 	return cmd
 }

command.go

@@ -298,6 +298,42 @@ func writeCmd(wr *proto.Writer, cmd Cmder) error {
 // cmdFirstKeyPosWithInfo returns the first key position in a command's args (0 if none).
 // Uses CommandInfo.FirstKeyPos when available (via cache peek, no network call), falling
 // back to a hardcoded table. eval/evalsha variants are resolved from the runtime numkeys arg.
+// cmdArgAfterToken returns the position after the first argument, from
+// position from on, that equals token (case-insensitive), or 0 when there is
+// none or nothing follows it.
+func cmdArgAfterToken(cmd Cmder, from int, token string) int {
+	n := len(cmd.Args())
+	for i := from; i < n-1; i++ {
+		if strings.EqualFold(cmd.stringArg(i), token) {
+			return i + 1
+		}
+	}
+	return 0
+}
+
+// migrateKeysPos returns the position of the first key of a MIGRATE KEYS
+// clause, walking the options so an operand (an AUTH password that reads
+// "keys") is not taken for the clause. 0 when there is no key.
+func migrateKeysPos(cmd Cmder) int {
+	n := len(cmd.Args())
+	for i := 6; i < n; {
+		switch strings.ToLower(cmd.stringArg(i)) {
+		case "keys":
+			if i+1 < n {
+				return i + 1
+			}
+			return 0
+		case "auth":
+			i += 2 // AUTH password
+		case "auth2":
+			i += 3 // AUTH2 username password
+		default:
+			i++ // COPY, REPLACE
+		}
+	}
+	return 0
+}
+
 func cmdFirstKeyPosWithInfo(cmd Cmder, info *CommandInfo) int {
 	if pos := cmd.firstKeyPos(); pos != 0 {
 		return int(pos)
@@ -319,7 +355,9 @@ func cmdFirstKeyPosWithInfo(cmd Cmder, info *CommandInfo) int {
 	}
 
 	switch name {
-	case "eval", "evalsha", "eval_ro", "evalsha_ro":
+	// FCALL has the EVAL layout: name, function, numkeys, keys... The typed
+	// FCall sets its key position itself; this covers raw Do/Process calls.
+	case "eval", "evalsha", "eval_ro", "evalsha_ro", "fcall", "fcall_ro":
 		if cmd.stringArg(2) != "0" {
 			return 3
 		}
@@ -335,6 +373,37 @@ func cmdFirstKeyPosWithInfo(cmd Cmder, info *CommandInfo) int {
 		// fallback only covers raw Do("msetex", ...) calls, which aren't
 		// guaranteed to route correctly and aren't the recommended usage.
 		return 2
+
+	// Raw forms of the commands below. The typed constructors set the key
+	// position; a raw Do / NewCmd does not, and args[1] is not a key.
+	case "xread":
+		// XREAD [COUNT n] [BLOCK ms] STREAMS key... id...
+		return cmdArgAfterToken(cmd, 1, "streams")
+	case "xreadgroup":
+		// XREADGROUP GROUP group consumer [...] STREAMS key... id...; the
+		// scan starts after the group and consumer names.
+		return cmdArgAfterToken(cmd, 4, "streams")
+	case "object", "xinfo":
+		// OBJECT|XINFO subcommand key; HELP has no key.
+		if len(cmd.Args()) > 2 {
+			return 2
+		}
+		return 0
+	case "bitop":
+		// BITOP op destkey key...
+		return 2
+	case "lmpop", "zmpop", "sintercard", "zintercard":
+		// numkeys key...
+		return 2
+	case "blmpop", "bzmpop":
+		// timeout numkeys key...
+		return 3
+	case "migrate":
+		// MIGRATE host port key|"" db timeout [...] [KEYS key...]
+		if cmd.stringArg(3) != "" {
+			return 3
+		}
+		return migrateKeysPos(cmd)
 	}
 
 	// Use CommandInfo cache when warm (in-memory only, no extra round-trips).

command_keypos_migrate_test.go

@@ -0,0 +1,44 @@
+package redis
+
+import (
+	"context"
+	"testing"
+)
+
+// Raw MIGRATE in its KEYS form: the options before KEYS take operands (AUTH
+// password, AUTH2 username password), so a scan for the first "keys" token
+// took a password that happens to be "KEYS" for the clause and routed by the
+// token after it. The resolver walks the option grammar instead.
+func TestRawMigrateKeysSkipsOptionOperands(t *testing.T) {
+	ctx := context.Background()
+	const K = "KEY_SENTINEL"
+	cases := [][]interface{}{
+		{"migrate", "h", 1, "", 0, 1000, "keys", K},
+		{"migrate", "h", 1, "", 0, 1000, "copy", "replace", "keys", K},
+		{"migrate", "h", 1, "", 0, 1000, "AUTH", "KEYS", "KEYS", K},
+		{"migrate", "h", 1, "", 0, 1000, "auth2", "keys", "KEYS", "keys", K},
+		{"migrate", "h", 1, "", 0, 1000, "copy", "auth", "pw", "keys", K, "b"},
+	}
+	for _, args := range cases {
+		want := -1
+		for i := len(args) - 1; i >= 0; i-- {
+			if s, ok := args[i].(string); ok && s == K {
+				want = i
+				break
+			}
+		}
+		if pos := cmdFirstKeyPosWithInfo(NewCmd(ctx, args...), nil); pos != want {
+			t.Errorf("%v: resolver picks position %d, key is at %d", args, pos, want)
+		}
+	}
+	// No KEYS clause, or nothing after it: keyless.
+	for _, args := range [][]interface{}{
+		{"migrate", "h", 1, "", 0, 1000},
+		{"migrate", "h", 1, "", 0, 1000, "auth", "keys"},
+		{"migrate", "h", 1, "", 0, 1000, "keys"},
+	} {
+		if pos := cmdFirstKeyPosWithInfo(NewCmd(ctx, args...), nil); pos != 0 {
+			t.Errorf("%v: resolver picks position %d, want 0", args, pos)
+		}
+	}
+}

command_keypos_raw_test.go

@@ -0,0 +1,54 @@
+package redis
+
+import (
+	"context"
+	"testing"
+)
+
+// Raw forms (Do / NewCmd) of the commands whose key is not args[1] carry no
+// declared key position, so the shared resolver must find the key itself.
+// Without that it fell back to args[1]: raw XREAD hashed "streams" and raw
+// LMPOP hashed its numkeys, which routes to the wrong Ring shard, the wrong
+// full-duplex engine, or a cold cluster slot.
+func TestRawCommandsRouteByTheirKey(t *testing.T) {
+	ctx := context.Background()
+	const K = "KEY_SENTINEL"
+	cases := [][]interface{}{
+		{"xread", "streams", K, "0"},
+		{"XREAD", "COUNT", 1, "STREAMS", K, "0"},
+		{"xread", "count", 1, "block", 0, "streams", K, "b", "0", "0"},
+		{"xreadgroup", "group", "g", "c", "streams", K, ">"},
+		{"xreadgroup", "group", "streams", "streams", "count", 1, "noack", "streams", K, ">"},
+		{"object", "encoding", K},
+		{"xinfo", "stream", K},
+		{"bitop", "and", K, "b"},
+		{"lmpop", 1, K, "left"},
+		{"zmpop", 1, K, "min"},
+		{"sintercard", 1, K},
+		{"zintercard", 2, K, "b"},
+		{"blmpop", 0, 1, K, "left"},
+		{"bzmpop", 0, 1, K, "min"},
+		{"migrate", "h", 1, K, 0, 1000},
+		{"migrate", "h", 1, "", 0, 1000, "copy", "keys", K, "b"},
+	}
+	for _, args := range cases {
+		cmd := NewCmd(ctx, args...)
+		want := -1
+		for i, a := range args {
+			if s, ok := a.(string); ok && s == K {
+				want = i
+				break
+			}
+		}
+		if pos := cmdFirstKeyPosWithInfo(cmd, nil); pos != want {
+			t.Errorf("%v: resolver picks position %d, key is at %d", args, pos, want)
+		}
+	}
+
+	// Subcommand help has no key.
+	for _, args := range [][]interface{}{{"object", "help"}, {"xinfo", "help"}} {
+		if pos := cmdFirstKeyPosWithInfo(NewCmd(ctx, args...), nil); pos != 0 {
+			t.Errorf("%v: resolver picks position %d, want 0 (keyless)", args, pos)
+		}
+	}
+}

command_keypos_test.go

@@ -0,0 +1,85 @@
+package redis
+
+// Typed commands whose key is not args[1] must say where it is, because the
+// shared resolver (cmdFirstKeyPosWithInfo) routes by it: Ring shard selection,
+// cluster slot routing and full-duplex engine routing. Each case builds the
+// command with a sentinel key and checks the resolver lands on it.
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func TestTypedCommandsRouteByTheirKey(t *testing.T) {
+	ctx := context.Background()
+	var got Cmder
+	c := cmdable(func(_ context.Context, cmd Cmder) error { got = cmd; return nil })
+	const K = "KEY_SENTINEL"
+	cases := []struct {
+		name string
+		fn   func()
+	}{
+		{"ObjectEncoding", func() { c.ObjectEncoding(ctx, K) }},
+		{"ObjectRefCount", func() { c.ObjectRefCount(ctx, K) }},
+		{"ObjectIdleTime", func() { c.ObjectIdleTime(ctx, K) }},
+		{"ObjectFreq", func() { c.ObjectFreq(ctx, K) }},
+		{"MemoryUsage", func() { c.MemoryUsage(ctx, K) }},
+		{"XInfoStream", func() { c.XInfoStream(ctx, K) }},
+		{"XInfoStreamFull", func() { c.XInfoStreamFull(ctx, K, 0) }},
+		{"XInfoGroups", func() { c.XInfoGroups(ctx, K) }},
+		{"XInfoConsumers", func() { c.XInfoConsumers(ctx, K, "g") }},
+		{"XGroupCreate", func() { c.XGroupCreate(ctx, K, "g", "$") }},
+		{"XGroupCreateMkStream", func() { c.XGroupCreateMkStream(ctx, K, "g", "$") }},
+		{"XGroupDestroy", func() { c.XGroupDestroy(ctx, K, "g") }},
+		{"XGroupSetID", func() { c.XGroupSetID(ctx, K, "g", "$") }},
+		{"XGroupCreateConsumer", func() { c.XGroupCreateConsumer(ctx, K, "g", "c") }},
+		{"XGroupDelConsumer", func() { c.XGroupDelConsumer(ctx, K, "g", "c") }},
+		{"XRead", func() { c.XRead(ctx, &XReadArgs{Streams: []string{K, "0"}}) }},
+		{"XReadGroup", func() { c.XReadGroup(ctx, &XReadGroupArgs{Group: "g", Consumer: "c", Streams: []string{K, "0"}}) }},
+		{"ZUnion", func() { c.ZUnion(ctx, ZStore{Keys: []string{K, "b"}}) }},
+		{"ZInter", func() { c.ZInter(ctx, &ZStore{Keys: []string{K, "b"}}) }},
+		{"ZDiff", func() { c.ZDiff(ctx, K, "b") }},
+		{"ZInterCard", func() { c.ZInterCard(ctx, 0, K, "b") }},
+		{"SInterCard", func() { c.SInterCard(ctx, 0, K, "b") }},
+		{"LMPop", func() { c.LMPop(ctx, "left", 1, K) }},
+		{"BLMPop", func() { c.BLMPop(ctx, 0, "left", 1, K) }},
+		{"ZMPop", func() { c.ZMPop(ctx, "min", 1, K) }},
+		{"BZMPop", func() { c.BZMPop(ctx, 0, "min", 1, K) }},
+		{"BitOpAnd", func() { c.BitOpAnd(ctx, K, "b") }},
+		{"Migrate", func() { c.Migrate(ctx, "h", "1", K, 0, time.Second) }},
+		{"Eval", func() { c.Eval(ctx, "s", []string{K}) }},
+		{"EvalSha", func() { c.EvalSha(ctx, "s", []string{K}) }},
+		{"EvalRO", func() { c.EvalRO(ctx, "s", []string{K}) }},
+		{"FCall", func() { c.FCall(ctx, "f", []string{K}) }},
+		{"FCallRO", func() { c.FCallRo(ctx, "f", []string{K}) }},
+		{"Sort", func() { c.Sort(ctx, K, &Sort{}) }},
+		{"GeoRadius", func() { c.GeoRadius(ctx, K, 0, 0, &GeoRadiusQuery{Radius: 1}) }},
+		{"XAutoClaim", func() { c.XAutoClaim(ctx, &XAutoClaimArgs{Stream: K, Group: "g", Consumer: "c", Start: "0"}) }},
+		{"Copy", func() { c.Copy(ctx, K, "b", 0, false) }},
+		{"LCS", func() { c.LCS(ctx, &LCSQuery{Key1: K, Key2: "b"}) }},
+	}
+	for _, tc := range cases {
+		got = nil
+		tc.fn()
+		if got == nil {
+			t.Errorf("%s: no command captured", tc.name)
+			continue
+		}
+		args := got.Args()
+		want := -1
+		for i, a := range args {
+			if s, ok := a.(string); ok && s == K {
+				want = i
+				break
+			}
+		}
+		if pos := cmdFirstKeyPosWithInfo(got, nil); pos != want {
+			at := "<none>"
+			if pos > 0 && pos < len(args) {
+				at = got.stringArg(pos)
+			}
+			t.Errorf("%s: resolver picks position %d (%q), key is at %d: %v", tc.name, pos, at, want, args)
+		}
+	}
+}

generic_commands.go

@@ -154,6 +154,7 @@ func (c cmdable) Migrate(ctx context.Context, host, port, key string, db int, ti
 		db,
 		formatMs(ctx, timeout),
 	)
+	cmd.SetFirstKeyPos(3) // the key is not args[1]: route by it
 	cmd.setReadTimeout(timeout)
 	_ = c(ctx, cmd)
 	return cmd
@@ -167,24 +168,28 @@ func (c cmdable) Move(ctx context.Context, key string, db int) *BoolCmd {
 
 func (c cmdable) ObjectFreq(ctx context.Context, key string) *IntCmd {
 	cmd := NewIntCmd(ctx, "object", "freq", key)
+	cmd.SetFirstKeyPos(2) // the key is not args[1]: route by it
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) ObjectRefCount(ctx context.Context, key string) *IntCmd {
 	cmd := NewIntCmd(ctx, "object", "refcount", key)
+	cmd.SetFirstKeyPos(2) // the key is not args[1]: route by it
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) ObjectEncoding(ctx context.Context, key string) *StringCmd {
 	cmd := NewStringCmd(ctx, "object", "encoding", key)
+	cmd.SetFirstKeyPos(2) // the key is not args[1]: route by it
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) ObjectIdleTime(ctx context.Context, key string) *DurationCmd {
 	cmd := NewDurationCmd(ctx, time.Second, "object", "idletime", key)
+	cmd.SetFirstKeyPos(2) // the key is not args[1]: route by it
 	_ = c(ctx, cmd)
 	return cmd
 }

list_commands.go

@@ -61,6 +61,10 @@ func (c cmdable) BLMPop(ctx context.Context, timeout time.Duration, direction st
 	}
 	args = append(args, strings.ToLower(direction), "count", count)
 	cmd := NewKeyValuesCmd(ctx, args...)
+	// The key follows numkeys, not args[1]: route by it.
+	if len(keys) > 0 {
+		cmd.SetFirstKeyPos(3)
+	}
 	cmd.setReadTimeout(timeout)
 	_ = c(ctx, cmd)
 	return cmd
@@ -114,6 +118,10 @@ func (c cmdable) LMPop(ctx context.Context, direction string, count int64, keys
 	}
 	args = append(args, strings.ToLower(direction), "count", count)
 	cmd := NewKeyValuesCmd(ctx, args...)
+	// The key follows numkeys, not args[1]: route by it.
+	if len(keys) > 0 {
+		cmd.SetFirstKeyPos(2)
+	}
 	_ = c(ctx, cmd)
 	return cmd
 }

set_commands.go

@@ -172,6 +172,10 @@ func (c cmdable) SInterCard(ctx context.Context, limit int64, keys ...string) *I
 	args[2+numKeys] = "limit"
 	args[3+numKeys] = limit
 	cmd := NewIntCmd(ctx, args...)
+	// The key follows numkeys, not args[1]: route by it.
+	if len(keys) > 0 {
+		cmd.SetFirstKeyPos(2)
+	}
 	_ = c(ctx, cmd)
 	return cmd
 }

sortedset_commands.go

@@ -108,6 +108,10 @@ func (c cmdable) BZMPop(ctx context.Context, timeout time.Duration, order string
 	}
 	args = append(args, strings.ToLower(order), "count", count)
 	cmd := NewZSliceWithKeyCmd(ctx, args...)
+	// The key follows numkeys, not args[1]: route by it.
+	if len(keys) > 0 {
+		cmd.SetFirstKeyPos(3)
+	}
 	cmd.setReadTimeout(timeout)
 	_ = c(ctx, cmd)
 	return cmd
@@ -270,6 +274,10 @@ func (c cmdable) ZInterCard(ctx context.Context, limit int64, keys ...string) *I
 	args[2+numKeys] = "limit"
 	args[3+numKeys] = limit
 	cmd := NewIntCmd(ctx, args...)
+	// The key follows numkeys, not args[1]: route by it.
+	if len(keys) > 0 {
+		cmd.SetFirstKeyPos(2)
+	}
 	_ = c(ctx, cmd)
 	return cmd
 }
@@ -286,6 +294,10 @@ func (c cmdable) ZMPop(ctx context.Context, order string, count int64, keys ...s
 	}
 	args = append(args, strings.ToLower(order), "count", count)
 	cmd := NewZSliceWithKeyCmd(ctx, args...)
+	// The key follows numkeys, not args[1]: route by it.
+	if len(keys) > 0 {
+		cmd.SetFirstKeyPos(2)
+	}
 	_ = c(ctx, cmd)
 	return cmd
 }

stream_commands.go

@@ -693,18 +693,21 @@ func (c cmdable) XTrimMinIDApproxMode(ctx context.Context, key string, minID str
 
 func (c cmdable) XInfoConsumers(ctx context.Context, key string, group string) *XInfoConsumersCmd {
 	cmd := NewXInfoConsumersCmd(ctx, key, group)
+	cmd.SetFirstKeyPos(2) // the key is not args[1]: route by it
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) XInfoGroups(ctx context.Context, key string) *XInfoGroupsCmd {
 	cmd := NewXInfoGroupsCmd(ctx, key)
+	cmd.SetFirstKeyPos(2) // the key is not args[1]: route by it
 	_ = c(ctx, cmd)
 	return cmd
 }
 
 func (c cmdable) XInfoStream(ctx context.Context, key string) *XInfoStreamCmd {
 	cmd := NewXInfoStreamCmd(ctx, key)
+	cmd.SetFirstKeyPos(2) // the key is not args[1]: route by it
 	_ = c(ctx, cmd)
 	return cmd
 }
@@ -718,6 +721,7 @@ func (c cmdable) XInfoStreamFull(ctx context.Context, key string, count int) *XI
 		args = append(args, "count", count)
 	}
 	cmd := NewXInfoStreamFullCmd(ctx, args...)
+	cmd.SetFirstKeyPos(2) // the key is not args[1]: route by it
 	_ = c(ctx, cmd)
 	return cmd
 }

unit_test.go

@@ -75,6 +75,26 @@ func TestCmdFirstKeyPosWithInfo_UsesCommandInfoWhenWarm(t *testing.T) {
 	}
 }
 
+// TestCmdFirstKeyPosWithInfo_RawFCall checks that a raw FCALL/FCALL_RO, which
+// has no key position set by a constructor, resolves its key after numkeys
+// (position 3) the way EVAL does, instead of hashing the function name. Ring
+// shard selection, cluster slot routing and full-duplex engine routing all
+// use this resolver.
+func TestCmdFirstKeyPosWithInfo_RawFCall(t *testing.T) {
+	ctx := context.Background()
+	for _, name := range []string{"fcall", "fcall_ro"} {
+		if got := cmdFirstKeyPosWithInfo(NewCmd(ctx, name, "myfn", "1", "k", "arg"), nil); got != 3 {
+			t.Fatalf("%s with one key: got %d, want 3", name, got)
+		}
+		if got := cmdFirstKeyPosWithInfo(NewCmd(ctx, name, "myfn", 1, "k"), nil); got != 3 {
+			t.Fatalf("%s with an int numkeys: got %d, want 3", name, got)
+		}
+		if got := cmdFirstKeyPosWithInfo(NewCmd(ctx, name, "myfn", "0", "arg"), nil); got != 0 {
+			t.Fatalf("%s with no keys: got %d, want 0 (keyless)", name, got)
+		}
+	}
+}
+
 // TestCmdFirstKeyPosWithInfo_PolicyTableKeyless checks that a module command
 // registered keyless in the static policy table routes as keyless even on a
 // cold command-info cache, and that key-carrying or specially-routed module