go-redis/command_metadata_test.go

package redis

import (
	"context"
	"errors"
	"fmt"
	"math"
	"net"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/redis/go-redis/v9/internal/pool"
	"github.com/redis/go-redis/v9/internal/proto"
	"github.com/redis/go-redis/v9/internal/routing"
)

func (s *commandMetadataStore) refreshOnce(ctx context.Context) error {
	return s.refreshMetadata(ctx, commandMetadataVerify)
}

func waitForCommandMetadataRefresh(t *testing.T, s *commandMetadataStore) {
	t.Helper()
	s.mu.Lock()
	done := s.refreshDone
	s.mu.Unlock()
	if done == nil {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	select {
	case <-done:
	case <-ctx.Done():
		t.Fatal("metadata refresh did not finish")
	}
}

func testReadOnlyCommandInfo(name string) *CommandInfo {
	return &CommandInfo{
		Name: name, Flags: []string{"readonly"}, FirstKeyPos: 1, LastKeyPos: 1, StepCount: 1,
		KeySpecs: []KeySpec{{Flags: []string{"RO"}, BeginSearch: "index", Index: 1, FindKeys: "range", KeyStep: 1}},
	}
}

func testCommandMetadataFetchResult(records map[string]*CommandInfo) commandMetadataFetchResult {
	return commandMetadataFetchResult{
		records:           records,
		serverName:        "redis",
		serverVersion:     "8.10.0",
		serverFingerprint: "8.10.0",
	}
}

func testCommandMetadataFetchResultFor(
	records map[string]*CommandInfo,
	version, fingerprint string,
) commandMetadataFetchResult {
	return commandMetadataFetchResult{
		records:           records,
		serverName:        "redis",
		serverVersion:     version,
		serverFingerprint: fingerprint,
	}
}

func waitForCondition(t *testing.T, timeout time.Duration, cond func() bool) bool {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		if cond() {
			return true
		}
		time.Sleep(5 * time.Millisecond)
	}
	return cond()
}

func TestCommandMetadataStaticStoreNeverStartsWorker(t *testing.T) {
	s := newCommandMetadataStore(&CommandMetadataConfig{
		Overrides: map[string]*CommandInfo{"get": nil},
	}, func(context.Context) (commandMetadataFetchResult, error) {
		t.Error("static mode must never fetch")
		return commandMetadataFetchResult{}, nil
	})
	s.onConnInit()
	s.requestRefresh()
	s.requestVerification()
	s.onClusterServerHelloReply(map[string]interface{}{"version": "8.10.0"})
	s.onClusterServerHelloReply(nil)
	s.mu.Lock()
	started := s.started
	s.mu.Unlock()
	if started {
		t.Error("static store must not start a worker")
	}
	s.stopAndJoin() // must not hang on a never-started worker
}

func TestCommandMetadataDynamicResolverActivatesTrackingLazily(t *testing.T) {
	var fetches atomic.Int32
	s := newCommandMetadataStoreForLive(&CommandMetadataConfig{
		RefreshInterval: 5 * time.Millisecond,
	}, func(context.Context) (commandMetadataFetchResult, error) {
		fetches.Add(1)
		return testCommandMetadataFetchResult(nil), nil
	})
	defer s.stopAndJoin()
	dynamic := newCommandMetadataPolicyResolverWithEnsure(s.view, s.ensureLive)
	custom := NewCommandInfoResolver(func(context.Context, Cmder) *routing.CommandPolicy {
		return &routing.CommandPolicy{Request: routing.ReqDefault}
	})
	custom.SetFallbackResolver(dynamic)
	ctx := context.Background()
	cmd := NewStringCmd(ctx, "get", "key")
	if policy := custom.GetCommandPolicy(ctx, cmd); policy == nil {
		t.Fatal("custom resolver did not supply its policy")
	}
	s.onClusterServerHelloReply(map[string]interface{}{"version": "8.10.0"})
	s.onClusterServerHelloReply(nil)
	s.onConnInit()
	s.mu.Lock()
	started := s.started
	s.mu.Unlock()
	if started || s.wantsServerHello() || fetches.Load() != 0 {
		t.Fatal("unused dynamic fallback activated metadata tracking")
	}

	if policy := dynamic.GetCommandPolicy(ctx, cmd); policy == nil || !s.view().live {
		t.Fatal("dynamic lookup did not synchronously load metadata")
	}
	if !s.wantsServerHello() {
		t.Fatal("dynamic lookup did not enable identity tracking")
	}
	before := fetches.Load()
	if !waitForCondition(t, time.Second, func() bool { return fetches.Load() > before }) {
		t.Fatal("dynamic lookup did not activate the configured periodic refresh")
	}
}

func TestCommandMetadataFetchErrorRetries(t *testing.T) {
	oldMin := cmdMetaBackoffMin
	cmdMetaBackoffMin = time.Millisecond
	defer func() { cmdMetaBackoffMin = oldMin }()

	var calls atomic.Int32
	s := newCommandMetadataStore(&CommandMetadataConfig{Mode: CommandMetadataPreferLive},
		func(context.Context) (commandMetadataFetchResult, error) {
			if calls.Add(1) < 3 {
				return commandMetadataFetchResult{}, errors.New("transient dial failure")
			}
			return testCommandMetadataFetchResult(nil), nil
		})
	defer s.stopAndJoin()

	s.onConnInit()
	if !waitForCondition(t, 5*time.Second, func() bool { return s.view().live }) {
		t.Fatalf("refresh never recovered after transient errors (%d calls)", calls.Load())
	}
}

func TestCommandMetadataPeriodicRefreshRetriesWhileLive(t *testing.T) {
	oldMin := cmdMetaBackoffMin
	cmdMetaBackoffMin = time.Millisecond
	defer func() { cmdMetaBackoffMin = oldMin }()

	var calls atomic.Int32
	periodicFailed := make(chan struct{})
	s := newCommandMetadataStore(&CommandMetadataConfig{
		Mode:            CommandMetadataPreferLive,
		RefreshInterval: 500 * time.Millisecond,
	}, func(context.Context) (commandMetadataFetchResult, error) {
		switch calls.Add(1) {
		case 1:
			return testCommandMetadataFetchResult(nil), nil
		case 2:
			close(periodicFailed)
			return commandMetadataFetchResult{}, errors.New("transient periodic refresh failure")
		default:
			return testCommandMetadataFetchResult(nil), nil
		}
	})
	defer s.stopAndJoin()

	s.onConnInit()
	if !waitForCondition(t, 5*time.Second, func() bool { return s.view().live }) {
		t.Fatal("initial live view was never published")
	}
	select {
	case <-periodicFailed:
	case <-time.After(2 * time.Second):
		t.Fatal("periodic refresh never ran")
	}
	if !waitForCondition(t, 100*time.Millisecond, func() bool { return calls.Load() >= 3 }) {
		t.Fatalf("failed periodic refresh was not retried promptly (%d calls)", calls.Load())
	}
}

func TestCommandMetadataRefreshReusesUnchangedView(t *testing.T) {
	for _, tc := range []struct {
		name   string
		change func(*commandMetadataFetchResult)
	}{
		{name: "unchanged"},
		{name: "routing only", change: func(m *commandMetadataFetchResult) {
			m.records["dbsize"].Tips = []string{"request_policy:all_shards", "response_policy:agg_sum"}
		}},
		{name: "key spec", change: func(m *commandMetadataFetchResult) {
			m.records["get"].KeySpecs[0].Index = 2
		}},
		{name: "tombstone", change: func(m *commandMetadataFetchResult) {
			m.records["get"] = nil
		}},
		{name: "removed record", change: func(m *commandMetadataFetchResult) {
			delete(m.records, "get")
		}},
		{name: "legacy provenance", change: func(m *commandMetadataFetchResult) {
			m.legacyRecords = map[string]struct{}{"dbsize": {}}
		}},
		{name: "server version", change: func(m *commandMetadataFetchResult) {
			m.serverVersion = "7.4.0"
		}},
		{name: "server name", change: func(m *commandMetadataFetchResult) {
			m.serverName = "other"
		}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			calls := 0
			s := newCommandMetadataStoreForLive(nil, func(context.Context) (commandMetadataFetchResult, error) {
				// Each fetch owns distinct records, as separately parsed replies do.
				m := testCommandMetadataFetchResult(map[string]*CommandInfo{
					"get":    testReadOnlyCommandInfo("get"),
					"dbsize": {Name: "dbsize", Arity: 1, Flags: []string{"readonly"}},
				})
				if calls > 0 && tc.change != nil {
					tc.change(&m)
				}
				calls++
				return m, nil
			})
			defer s.stopAndJoin()
			if err := s.ensureLive(context.Background()); err != nil {
				t.Fatal(err)
			}
			first := s.view()
			if err := s.refreshOnce(context.Background()); err != nil {
				t.Fatal(err)
			}
			second := s.view()
			if changed := second != first; changed != (tc.change != nil) {
				t.Fatalf("view changed = %v, want %v", changed, tc.change != nil)
			}
			if tc.name == "routing only" && first.cscFingerprint != second.cscFingerprint {
				t.Fatal("routing-only fixture unexpectedly changed CSC eligibility")
			}
			if err := s.refreshOnce(context.Background()); err != nil {
				t.Fatal(err)
			}
			if s.view() != second {
				t.Fatal("identical refresh rebuilt the live view")
			}
		})
	}
}

func TestCommandMetadataUnchangedRefreshValidatesIdentity(t *testing.T) {
	for _, tc := range []struct {
		name   string
		change func(*commandMetadataStore, *commandMetadataFetchResult)
	}{
		{"missing identity", func(_ *commandMetadataStore, m *commandMetadataFetchResult) {
			m.serverFingerprint = ""
		}},
		{"different identity", func(_ *commandMetadataStore, m *commandMetadataFetchResult) {
			m.serverFingerprint = "other-server"
		}},
		{"invalidated during fetch", func(s *commandMetadataStore, _ *commandMetadataFetchResult) {
			s.invalidateLive()
		}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			var duringFetch func(*commandMetadataFetchResult)
			s := newCommandMetadataStoreForLive(nil, func(context.Context) (commandMetadataFetchResult, error) {
				m := testCommandMetadataFetchResult(map[string]*CommandInfo{"get": testReadOnlyCommandInfo("get")})
				if duringFetch != nil {
					duringFetch(&m)
				}
				return m, nil
			})
			defer s.stopAndJoin()
			if err := s.ensureLive(context.Background()); err != nil {
				t.Fatal(err)
			}
			duringFetch = func(m *commandMetadataFetchResult) { tc.change(s, m) }
			if err := s.refreshOnce(context.Background()); err == nil {
				t.Fatal("unchanged records bypassed identity validation")
			}
			previous := s.view()
			duringFetch = nil
			if err := s.refreshOnce(context.Background()); err != nil {
				t.Fatal(err)
			}
			if !s.view().live {
				t.Fatal("valid refresh did not restore live metadata")
			}
			if previous.live && s.view() != previous {
				t.Fatal("rejected fetch replaced the last successful comparison input")
			}
		})
	}
}

func TestCommandMetadataNormalizesEffectiveRecords(t *testing.T) {
	upper := &CommandInfo{
		Name:  "MODULE.GET",
		Flags: []string{"READONLY", "Future_Flag"},
		Tips: []string{
			"REQUEST_POLICY:ALL_SHARDS",
			"RESPONSE_POLICY:AGG_SUM",
			"Future_Tip:MiXeD",
		},
		FirstKeyPos: 1, LastKeyPos: 1, StepCount: 1,
		KeySpecs: []KeySpec{{
			Flags:       []string{"ro", "ACCESS", "PREFIX", "Future_Key_Flag"},
			BeginSearch: "INDEX",
			Index:       1,
			FindKeys:    "RANGE",
			KeyStep:     1,
		}},
	}
	lower := cloneCommandInfo(upper)
	lower.Name = "module.get"
	lower.Flags[0] = "readonly"
	lower.Tips[0] = "request_policy:all_shards"
	lower.Tips[1] = "response_policy:agg_sum"
	lower.KeySpecs[0].Flags[0] = "RO"
	lower.KeySpecs[0].Flags[1] = "access"
	lower.KeySpecs[0].Flags[2] = "prefix"
	lower.KeySpecs[0].BeginSearch = "index"
	lower.KeySpecs[0].FindKeys = "range"

	upperView := buildCommandMetadataView(nil, map[string]*CommandInfo{"MODULE.GET": upper})
	lowerView := buildCommandMetadataView(nil, map[string]*CommandInfo{"module.get": lower})
	if upperView.cscFingerprint != lowerView.cscFingerprint {
		t.Fatal("equivalent normalized records produced different CSC decisions")
	}
	record := upperView.records["module.get"]
	if record == nil || record.Name != "module.get" || record.Flags[0] != "readonly" ||
		record.Tips[0] != "request_policy:all_shards" ||
		record.Tips[1] != "response_policy:agg_sum" ||
		record.KeySpecs[0].Flags[0] != "RO" || record.KeySpecs[0].Flags[1] != "access" ||
		record.KeySpecs[0].Flags[2] != "prefix" ||
		record.KeySpecs[0].BeginSearch != "index" || record.KeySpecs[0].FindKeys != "range" {
		t.Fatalf("effective record was not normalized: %+v", record)
	}
	if record.Flags[1] != "Future_Flag" || record.Tips[2] != "Future_Tip:MiXeD" ||
		record.KeySpecs[0].Flags[3] != "Future_Key_Flag" {
		t.Fatalf("unknown extensions were not preserved verbatim: %+v", record)
	}
	upperRouting := upperView.routingTable["module.get"]
	lowerRouting := lowerView.routingTable["module.get"]
	if !upperRouting.valid || !lowerRouting.valid ||
		upperRouting.policy.Request != lowerRouting.policy.Request ||
		upperRouting.policy.Response != lowerRouting.policy.Response ||
		upperRouting.keyState != lowerRouting.keyState {
		t.Fatalf("equivalent normalized records produced different routing metadata: upper=%+v lower=%+v", upperRouting, lowerRouting)
	}
}

func TestCommandMetadataLegacyShapedUnknownIsRoutingOnly(t *testing.T) {
	legacy := &CommandInfo{
		Name: "legacy.get", Flags: []string{"readonly"},
		FirstKeyPos: 1, LastKeyPos: 1, StepCount: 1,
	}
	view := buildCommandMetadataViewForServerWithLegacy(
		map[string]*CommandInfo{"LEGACY.GET": legacy},
		nil,
		"redis", "8.10.0",
		map[string]struct{}{"LEGACY.GET": {}},
	)
	cmd := makeCmd("legacy.get", "key")
	if isCacheableInView(view, cmd) {
		t.Fatal("live-only legacy-shaped record must not prove CSC eligibility")
	}
	if !commandRecordHas(view.records["legacy.get"], "dont_cache", true) {
		t.Fatal("legacy-shaped live-only record must carry a shared dont_cache correction")
	}
	meta, ok := routingLookupMeta(view, cmd)
	if !ok {
		t.Fatal("legacy-shaped record was not retained for routing")
	}
	if pos, ok := routingFirstKeyPos(meta, cmd); !ok || pos != 1 {
		t.Fatalf("legacy routing key = (%d, %v), want (1, true)", pos, ok)
	}
}

func TestCommandMetadataLegacyShapeUsesServerVersionCompatibility(t *testing.T) {
	legacyGet := &CommandInfo{
		Name: "get", Flags: []string{"readonly"},
		FirstKeyPos: 1, LastKeyPos: 1, StepCount: 1,
	}
	legacyTTL := &CommandInfo{
		Name: "ttl", Flags: []string{"readonly"},
		FirstKeyPos: 1, LastKeyPos: 1, StepCount: 1,
	}
	legacyXPending := &CommandInfo{
		Name: "xpending", Flags: []string{"readonly"},
		FirstKeyPos: 1, LastKeyPos: 1, StepCount: 1,
	}
	live := map[string]*CommandInfo{
		"get": legacyGet, "ttl": legacyTTL, "xpending": legacyXPending,
	}
	legacy := map[string]struct{}{"get": {}, "ttl": {}, "xpending": {}}

	pre810 := buildCommandMetadataViewForServerWithLegacy(
		live, nil, "redis", "6.2.0", legacy,
	)
	if !isCacheableInView(pre810, makeCmd("get", "key")) {
		t.Error("known pre-8.10 legacy record should use its legacy key positions")
	}
	if commandRecordHas(pre810.records["get"], "dont_cache", true) {
		t.Error("known pre-8.10 legacy record received an unnecessary dont_cache correction")
	}
	for _, name := range []string{"ttl", "xpending"} {
		if isCacheableInView(pre810, makeCmd(name, "key")) {
			t.Errorf("known pre-8.10 legacy %s lost its nondeterministic exclusion", name)
		}
		if !commandRecordHas(pre810.records[name], "nondeterministic_output", true) {
			t.Errorf("known pre-8.10 legacy %s did not retain its snapshot exclusion", name)
		}
	}

	redis810 := buildCommandMetadataViewForServerWithLegacy(
		live, nil, "redis", "8.10.0", legacy,
	)
	if isCacheableInView(redis810, makeCmd("get", "key")) {
		t.Error("legacy-shaped record from an 8.10+ server must fail closed for CSC")
	}
	if !commandRecordHas(redis810.records["get"], "dont_cache", true) {
		t.Error("8.10+ legacy inconsistency was not represented in the shared record")
	}
}

func TestCommandMetadataLegacyRoutingPolicies(t *testing.T) {
	for _, fields := range []int{6, 7, 10} {
		t.Run(fmt.Sprint(fields), func(t *testing.T) {
			var entries []string
			for _, name := range []string{"dbsize", "flushall", "mget"} {
				info := commandInfoSnapshotByName()[name]
				first, last, step := info.FirstKeyPos, info.LastKeyPos, info.StepCount
				if name == "mget" {
					first = 2 // Live positions must win over the snapshot's position 1.
				}
				entry := []string{
					commandInfoTestBulk(name), commandInfoTestInt(int64(info.Arity)),
					commandInfoTestArray(commandInfoTestBulk(info.Flags[0])),
					commandInfoTestInt(int64(first)), commandInfoTestInt(int64(last)), commandInfoTestInt(int64(step)),
				}
				for len(entry) < fields {
					entry = append(entry, commandInfoTestArray())
				}
				entries = append(entries, commandInfoTestArray(entry...))
			}
			parsed := commandInfoTestReadReply(t, commandInfoTestArray(entries...))
			view := buildCommandMetadataViewForServerWithLegacy(parsed.Val(), nil, "redis", "6.2.0", parsed.legacyRecords)
			for _, tc := range []struct {
				name     string
				request  routing.RequestPolicy
				response routing.ResponsePolicy
			}{
				{"dbsize", routing.ReqAllShards, routing.RespAggSum},
				{"flushall", routing.ReqAllShards, routing.RespAllSucceeded},
				{"mget", routing.ReqMultiShard, routing.RespDefaultHashSlot},
			} {
				policy := view.routingTable[tc.name].policy
				if fields == 10 {
					// Modern records explicitly supply tips, even when empty.
					if policy != nil && policy.Request != routing.ReqDefault {
						t.Errorf("%s: snapshot policy replaced modern live tips: %+v", tc.name, policy)
					}
				} else if policy == nil || policy.Request != tc.request || policy.Response != tc.response {
					t.Errorf("%s: policy=%+v, want %v/%v", tc.name, policy, tc.request, tc.response)
				}
			}
			plan, ok := routingResolveKeyPlan(view.routingTable["mget"], makeCmd("mget", "prefix", "{a}k", "{b}k"))
			if !ok || len(plan.positions) != 2 || plan.positions[0] != 2 || plan.positions[1] != 3 {
				t.Fatalf("live key plan=%+v, ok=%v", plan, ok)
			}
		})
	}
}

func TestCommandMetadataViewCopiesSourceRecords(t *testing.T) {
	for _, source := range []string{"live", "override"} {
		t.Run(source, func(t *testing.T) {
			live := &CommandInfo{
				Name:  "module.read",
				Flags: []string{"readonly"},
				Tips:  []string{"request_policy:all_shards"},
				KeySpecs: []KeySpec{{
					Flags: []string{"RO"}, BeginSearch: "index", Index: 1,
					FindKeys: "range", KeyStep: 1,
				}},
				CommandPolicy: &routing.CommandPolicy{
					Request: routing.ReqAllShards,
					Tips:    map[string]string{routing.ReadOnlyCMD: ""},
				},
			}
			view := buildCommandMetadataView(map[string]*CommandInfo{"module.read": live}, nil)
			if source == "override" {
				view = buildCommandMetadataView(nil, map[string]*CommandInfo{"module.read": live})
			}
			live.Flags[0] = "write"
			live.Tips[0] = "request_policy:all_nodes"
			live.KeySpecs[0].Index = 7
			live.KeySpecs[0].Flags[0] = "RW"
			live.CommandPolicy.Request = routing.ReqAllNodes
			delete(live.CommandPolicy.Tips, routing.ReadOnlyCMD)

			got := view.records["module.read"]
			if got == live || got.Flags[0] != "readonly" || got.Tips[0] != "request_policy:all_shards" ||
				got.KeySpecs[0].Index != 1 || got.KeySpecs[0].Flags[0] != "RO" ||
				got.CommandPolicy.Request != routing.ReqAllShards {
				t.Fatalf("live record was not deeply copied: %+v", got)
			}
			if _, ok := got.CommandPolicy.Tips[routing.ReadOnlyCMD]; !ok {
				t.Fatal("live CommandPolicy tips share the caller's map")
			}
		})
	}
}

func TestCommandMetadataLiveTombstonesBlockLowerLayers(t *testing.T) {
	view := buildCommandMetadataViewForServer(map[string]*CommandInfo{
		"GET":   nil,
		"TOUCH": nil,
	}, nil, "redis", "8.10.0")
	for _, name := range []string{"get", "touch"} {
		if _, ok := view.records[name]; ok {
			t.Errorf("live tombstone for %s exposed a lower-layer record", name)
		}
		if _, ok := view.tombstones[name]; !ok {
			t.Errorf("live tombstone for %s was not preserved in the view", name)
		}
		if _, ok := view.cscTable[name]; ok {
			t.Errorf("live tombstone for %s exposed a lower-layer CSC entry", name)
		}
	}

	keyedGet := testReadOnlyCommandInfo("myext.get")
	view = buildCommandMetadataViewForServer(
		map[string]*CommandInfo{"myext.get": nil},
		map[string]*CommandInfo{"MYEXT.GET": keyedGet},
		"redis", "7.4.0",
	)
	if !isCacheableInView(view, makeCmd("myext.get", "k")) {
		t.Error("the highest-priority application override did not replace a live tombstone")
	}
	if _, ok := view.tombstones["myext.get"]; ok {
		t.Error("a valid application override did not clear the lower live tombstone")
	}
}

func TestCommandMetadataTombstonedChildKeepsParentShadowed(t *testing.T) {
	view := buildCommandMetadataViewForServer(map[string]*CommandInfo{
		"container":       testReadOnlyCommandInfo("container"),
		"container|child": nil,
	}, nil, "redis", "8.10.0")

	cmd := makeCmd("container", "child", "key")
	if _, ok := cscLookupMeta(view, cmd); ok {
		t.Fatal("tombstoned child fell back to the bare parent for CSC")
	}
	if _, ok := routingLookupMeta(view, cmd); ok {
		t.Fatal("tombstoned child fell back to the bare parent for routing")
	}
	if _, ok := view.shadowedParents["container"]; !ok {
		t.Fatal("container parent was not kept explicitly shadowed")
	}
	if _, ok := view.tombstones["container|child"]; !ok {
		t.Fatal("the normalized child tombstone was not preserved")
	}
}

func TestCommandMetadataNormalizedCollisionsFailClosed(t *testing.T) {
	keyed := commandInfoSnapshotByName()["get"]
	for _, tc := range []struct {
		name            string
		live, overrides map[string]*CommandInfo
	}{
		{"live tombstone", map[string]*CommandInfo{"GET": nil, "get": keyed}, nil},
		{"live records", map[string]*CommandInfo{"GET": keyed, "get": keyed}, nil},
		{"overrides", nil, map[string]*CommandInfo{"GET": keyed, "get": keyed}},
		{"override tombstone", nil, map[string]*CommandInfo{"GET": nil}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			view := buildCommandMetadataView(tc.live, tc.overrides)
			if _, ok := view.records["get"]; ok {
				t.Fatal("ambiguous or tombstoned command exposed a record")
			}
			if _, ok := view.tombstones["get"]; !ok {
				t.Fatal("normalized tombstone was not preserved")
			}
			if isCacheableInView(view, makeCmd("get", "key")) {
				t.Fatal("ambiguous or tombstoned command enabled CSC")
			}
			if _, ok := routingLookupMeta(view, makeCmd("get", "key")); ok {
				t.Fatal("ambiguous or tombstoned command enabled routing")
			}
		})
	}
}

func TestCommandMetadataBareParentOverrideIsInert(t *testing.T) {
	// A bare parent override is pruned without hiding its subcommands.
	view := buildCommandMetadataView(nil, map[string]*CommandInfo{
		"memory": {
			Name: "memory", Flags: []string{"readonly"}, FirstKeyPos: 1, LastKeyPos: 1, StepCount: 1,
			KeySpecs: []KeySpec{{BeginSearch: "index", Index: 1, FindKeys: "range", KeyStep: 1}},
		},
	})
	if _, ok := view.cscTable["memory"]; ok {
		t.Error("bare container-parent override must be pruned from the table")
	}
	if isCacheableInView(view, makeCmd("memory", "usage", "k")) {
		t.Error("memory|usage must stay non-cacheable (dont_cache correction)")
	}
	meta, ok := cscLookupMeta(view, makeCmd("memory", "usage", "k"))
	if !ok || meta.bits&cscTipDontCache == 0 {
		t.Error("memory|usage must still resolve through the parent set")
	}
}

func TestCommandMetadataConcurrentRefreshAndStop(t *testing.T) {
	for i := 0; i < 20; i++ {
		s := newCommandMetadataStore(&CommandMetadataConfig{Mode: CommandMetadataPreferLive},
			func(context.Context) (commandMetadataFetchResult, error) {
				return testCommandMetadataFetchResult(nil), nil
			})
		var wg sync.WaitGroup
		for g := 0; g < 4; g++ {
			wg.Add(1)
			go func() {
				defer wg.Done()
				s.onConnInit()
			}()
		}
		wg.Add(1)
		go func() {
			defer wg.Done()
			s.stopAndJoin()
		}()
		wg.Wait()
		s.stopAndJoin()
	}
}

func TestCommandMetadataStopCancelsInflightFetch(t *testing.T) {
	// Close must not wait for a fetch that ignores its context.
	release := make(chan struct{})
	defer close(release)
	s := newCommandMetadataStore(&CommandMetadataConfig{Mode: CommandMetadataPreferLive},
		func(context.Context) (commandMetadataFetchResult, error) {
			<-release
			return commandMetadataFetchResult{}, errors.New("released late")
		})
	s.onConnInit()
	time.Sleep(20 * time.Millisecond) // let the worker enter the fetch
	done := make(chan struct{})
	go func() {
		s.stopAndJoin()
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("stopAndJoin blocked on an in-flight fetch")
	}
}

func TestCommandMetadataWaitingCallerCanCancel(t *testing.T) {
	for _, stopStore := range []bool{false, true} {
		t.Run(fmt.Sprintf("stop=%v", stopStore), func(t *testing.T) {
			started, release := make(chan struct{}), make(chan struct{})
			releaseFetch := sync.OnceFunc(func() { close(release) })
			defer releaseFetch()
			var fetches atomic.Int32
			store := newCommandMetadataStoreForLive(nil, func(context.Context) (commandMetadataFetchResult, error) {
				fetches.Add(1)
				close(started)
				<-release // Deliberately ignore cancellation, as a user hook might.
				return testCommandMetadataFetchResult(nil), nil
			})
			defer store.stopAndJoin()
			owner := make(chan error, 1)
			go func() { owner <- store.ensureLive(context.Background()) }()
			<-started

			ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
			defer cancel()
			if stopStore {
				ctx = context.Background()
			}
			waiter := make(chan error, 1)
			go func() { waiter <- store.ensureLive(ctx) }()
			if stopStore {
				store.stopAndJoin()
			}
			select {
			case err := <-waiter:
				want := context.DeadlineExceeded
				if stopStore {
					want = context.Canceled
				}
				if !errors.Is(err, want) {
					t.Fatalf("waiting caller returned %v, want %v", err, want)
				}
			case <-time.After(time.Second):
				t.Fatal("waiting caller could not cancel independently")
			}
			if !stopStore {
				select {
				case err := <-owner:
					t.Fatalf("waiting caller canceled the shared attempt: %v", err)
				default:
				}
				releaseFetch()
			}
			select {
			case err := <-owner:
				if stopStore && !errors.Is(err, context.Canceled) || !stopStore && err != nil {
					t.Fatalf("owning caller returned %v", err)
				}
			case <-time.After(time.Second):
				t.Fatal("owning attempt did not finish")
			}
			if fetches.Load() != 1 || store.view().live == stopStore {
				t.Fatalf("fetches=%d live=%v", fetches.Load(), store.view().live)
			}
		})
	}
}

func TestCommandMetadataWaitingCallerFallsBackAfterFailedAttempt(t *testing.T) {
	oldMin := cmdMetaBackoffMin
	cmdMetaBackoffMin = time.Millisecond
	defer func() { cmdMetaBackoffMin = oldMin }()
	for _, cancelOwner := range []bool{false, true} {
		t.Run(fmt.Sprintf("cancel=%v", cancelOwner), func(t *testing.T) {
			started, release := make(chan struct{}), make(chan struct{})
			retrying, recoverNode := make(chan struct{}), make(chan struct{})
			var fetches atomic.Int32
			store := newCommandMetadataStoreForLive(nil, func(ctx context.Context) (commandMetadataFetchResult, error) {
				if fetches.Add(1) == 1 {
					close(started)
					select {
					case <-ctx.Done():
						return commandMetadataFetchResult{}, ctx.Err()
					case <-release:
						return commandMetadataFetchResult{}, errors.New("temporary failure")
					}
				}
				close(retrying)
				select {
				case <-ctx.Done():
					return commandMetadataFetchResult{}, ctx.Err()
				case <-recoverNode:
					return testCommandMetadataFetchResult(nil), nil
				}
			})
			defer store.stopAndJoin()
			ctx, cancel := context.WithCancel(context.Background())
			defer cancel()
			owner, waiter := make(chan error, 1), make(chan error, 1)
			go func() { owner <- store.ensureLive(ctx) }()
			<-started
			go func() { waiter <- store.ensureLive(context.Background()) }()
			if cancelOwner {
				cancel()
			} else {
				close(release)
			}
			if err := <-owner; err == nil {
				t.Fatal("owning caller did not report its failure")
			}
			select {
			case err := <-waiter:
				if !errors.Is(err, errCommandMetadataNotReady) || store.view().live {
					t.Fatalf("waiting caller did not use conservative fallback: err=%v live=%v", err, store.view().live)
				}
			case <-time.After(time.Second):
				t.Fatal("waiting caller started another blocking probe")
			}
			select {
			case <-retrying:
			case <-time.After(time.Second):
				t.Fatal("background verification did not start")
			}
			for range 100 {
				ctx, cancel := context.WithTimeout(context.Background(), time.Second)
				err := store.ensureLive(ctx)
				cancel()
				if !errors.Is(err, errCommandMetadataNotReady) {
					t.Fatalf("command waited for background verification: %v", err)
				}
			}
			if store.view().live || fetches.Load() != 2 {
				t.Fatal("commands started extra probes or exposed unverified metadata")
			}
			close(recoverNode)
			if !waitForCondition(t, time.Second, func() bool { return store.view().live }) {
				t.Fatal("background recovery did not publish verified metadata")
			}
		})
	}
}

func TestCommandMetadataBackgroundRetryHonorsCooldown(t *testing.T) {
	oldMin, oldMax, oldCap := cmdMetaBackoffMin, cmdMetaBackoffMax, cmdMetaRetryCap
	cmdMetaBackoffMin, cmdMetaBackoffMax, cmdMetaRetryCap = 50*time.Millisecond, 100*time.Millisecond, 1
	defer func() { cmdMetaBackoffMin, cmdMetaBackoffMax, cmdMetaRetryCap = oldMin, oldMax, oldCap }()
	var mu sync.Mutex
	var attempts []time.Time
	var recovered atomic.Bool
	store := newCommandMetadataStoreForLive(nil, func(context.Context) (commandMetadataFetchResult, error) {
		mu.Lock()
		attempts = append(attempts, time.Now())
		mu.Unlock()
		if !recovered.Load() {
			return commandMetadataFetchResult{}, errors.New("replica verification failed")
		}
		return testCommandMetadataFetchResult(nil), nil
	})
	defer store.stopAndJoin()
	if err := store.ensureLive(context.Background()); err == nil {
		t.Fatal("initial verification unexpectedly succeeded")
	}
	if !waitForCondition(t, 2*time.Second, func() bool {
		// Command traffic must permit recovery after the retry cap, but cannot
		// bypass the cooldown between actual network attempts.
		_ = store.ensureLive(context.Background())
		mu.Lock()
		if len(attempts) >= 2 {
			recovered.Store(true)
		}
		mu.Unlock()
		return store.view().live
	}) {
		t.Fatal("commands did not trigger background recovery after the retry cap")
	}
	mu.Lock()
	defer mu.Unlock()
	for i := 1; i < len(attempts); i++ {
		if attempts[i].Sub(attempts[i-1]) < cmdMetaBackoffMin {
			t.Fatalf("command traffic bypassed cooldown: %v", attempts)
		}
	}
}

func TestCommandMetadataInvalidationDoesNotReopenForegroundWait(t *testing.T) {
	started, release := make(chan struct{}), make(chan struct{})
	var fetches atomic.Int32
	store := newCommandMetadataStoreForLive(nil, func(ctx context.Context) (commandMetadataFetchResult, error) {
		if fetches.Add(1) > 1 {
			close(started)
			select {
			case <-release:
			case <-ctx.Done():
				return commandMetadataFetchResult{}, ctx.Err()
			}
		}
		return testCommandMetadataFetchResult(nil), nil
	})
	defer store.stopAndJoin()
	if err := store.ensureLive(context.Background()); err != nil {
		t.Fatal(err)
	}
	store.invalidateLiveAndRequestRefresh()
	select {
	case <-started:
	case <-time.After(time.Second):
		t.Fatal("invalidation did not start background verification")
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := store.ensureLive(ctx); !errors.Is(err, errCommandMetadataNotReady) || store.view().live {
		t.Fatalf("invalidated view blocked a command or remained live: %v", err)
	}
	close(release)
	if !waitForCondition(t, time.Second, func() bool { return store.view().live }) {
		t.Fatal("verified replacement was not published")
	}
}

func TestCommandMetadataSilentNodeRespectsTimeoutAndCancellation(t *testing.T) {
	for _, command := range []string{"hello", "command"} {
		for _, cancelParent := range []bool{false, true} {
			t.Run(fmt.Sprintf("%s/cancel=%v", command, cancelParent), func(t *testing.T) {
				listener, err := net.Listen("tcp", "127.0.0.1:0")
				if err != nil {
					t.Fatal(err)
				}
				t.Cleanup(func() { _ = listener.Close() })
				stalled, release := make(chan struct{}), make(chan struct{})
				t.Cleanup(func() { close(release) })
				go func() {
					conn, err := listener.Accept()
					if err != nil {
						return
					}
					hellos := 0
					serveTestRESPConn(conn, func(name string) string {
						if name == "hello" {
							hellos++
						}
						// Allow connection initialization, then stop answering the probe.
						if name == command && (name != "hello" || hellos > 1) {
							close(stalled)
							<-release
						}
						switch name {
						case "hello":
							return "*2\r\n+version\r\n+8.10.0\r\n"
						case "command":
							return "*0\r\n"
						case "ping":
							return "+PONG\r\n"
						default:
							return "+OK\r\n"
						}
					})
				}()
				client := NewClient(&Options{
					Addr: listener.Addr().String(), Protocol: 2, DisableIdentity: true,
					MaxRetries: -1, PoolSize: 1, ReadTimeout: 50 * time.Millisecond,
				})
				t.Cleanup(func() { _ = client.Close() })
				if err := client.Ping(context.Background()).Err(); err != nil {
					t.Fatal(err)
				}
				store := newCommandMetadataStoreForLive(nil, client.baseClient.fetchCommandMetadata)
				t.Cleanup(store.stopAndJoin)
				ctx, cancel := context.WithTimeout(context.Background(), time.Second)
				defer cancel()
				started := time.Now()
				result := make(chan error, 1)
				go func() { result <- store.ensureLive(ctx) }()
				select {
				case <-stalled:
				case <-ctx.Done():
					t.Fatal("metadata probe never reached the server")
				}
				if cancelParent {
					cancel()
				}
				select {
				case err := <-result:
					if err == nil || store.view().live {
						t.Fatal("silent node published live metadata")
					}
					if cancelParent && !errors.Is(err, context.Canceled) {
						t.Fatalf("canceled fetch returned %v", err)
					}
					if time.Since(started) > 500*time.Millisecond {
						t.Fatal("one probe consumed the overall refresh budget")
					}
				case <-time.After(500 * time.Millisecond):
					t.Fatal("metadata probe ignored its timeout or cancellation")
				}
			})
		}
	}
}

func TestCommandMetadataFetchRejectsAndAdoptsDifferentServer(t *testing.T) {
	for _, tc := range []struct {
		name, server, version, previous string
	}{
		{"version changed", "redis", "8.10.0-A", "redis:8.10.0-B"},
		{"server changed", "other", "8.10.0", "redis:8.10.0"},
		{"server unknown", "", "8.10.0", "redis:8.10.0"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			fingerprint := tc.version
			hello := commandInfoTestMap(commandInfoTestBulk("version"), commandInfoTestBulk(tc.version))
			if tc.server != "" {
				fingerprint = tc.server + ":" + fingerprint
				hello = commandInfoTestMap(
					commandInfoTestBulk("server"), commandInfoTestBulk(tc.server),
					commandInfoTestBulk("version"), commandInfoTestBulk(tc.version),
				)
			}
			ln, err := net.Listen("tcp", "127.0.0.1:0")
			if err != nil {
				t.Fatalf("listen: %v", err)
			}
			go func() {
				conn, acceptErr := ln.Accept()
				if acceptErr != nil {
					return
				}
				serveTestRESPConn(conn, func(command string) string {
					switch command {
					case "hello":
						return hello
					case "command":
						return "*0\r\n"
					default:
						return "+OK\r\n"
					}
				})
			}()

			client := NewClient(&Options{
				Addr:            ln.Addr().String(),
				Protocol:        3,
				DisableIdentity: true,
				MaxRetries:      0,
			})
			store := newCommandMetadataStore(
				&CommandMetadataConfig{Mode: CommandMetadataPreferLive}, nil,
			)
			store.onServerHello(tc.previous)
			client.baseClient.cmdMeta = store
			t.Cleanup(func() {
				store.stopAndJoin()
				_ = client.Close()
				_ = ln.Close()
			})

			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
			defer cancel()
			if _, err := client.baseClient.fetchCommandMetadata(ctx); err == nil {
				t.Fatal("metadata fetched from a different server identity was accepted")
			}
			if got := store.serverFingerprint(); got != fingerprint {
				t.Fatalf("metadata fetch did not record the newly observed identity: %q", got)
			}

			metadata, err := client.baseClient.fetchCommandMetadata(ctx)
			if err != nil {
				t.Fatalf("metadata fetch from the target server failed: %v", err)
			}
			if len(metadata.records) != 0 {
				t.Fatalf("metadata fetch returned %d records, want 0", len(metadata.records))
			}
			if metadata.serverVersion != tc.version || metadata.serverName != tc.server {
				t.Fatalf("metadata server = %q %q, want %q %q", metadata.serverName, metadata.serverVersion, tc.server, tc.version)
			}
			if metadata.serverFingerprint != fingerprint {
				t.Fatalf("metadata server fingerprint = %q, want %q", metadata.serverFingerprint, fingerprint)
			}
		})
	}
}

func TestCommandMetadataFetchWithoutHello(t *testing.T) {
	for _, tc := range []struct {
		name         string
		infoReply    string
		moduleReply  string
		commandReply string
		version      string
		retry        bool
		wantErr      bool
	}{
		{name: "legacy module"},
		{name: "new version without server identity", version: "8.10.0", commandReply: commandInfoTestArray(commandInfoTestEntry10(
			"custom.read2", commandInfoTestArray(commandInfoTestBulk("readonly")), commandInfoTestArray(),
			commandInfoTestArray(commandInfoTestRangeKeySpec(commandInfoTestArray(commandInfoTestBulk("RO")), commandInfoTestBulk("index"), commandInfoTestInt(1))),
			commandInfoTestArray(),
		))},
		{name: "missing version", infoReply: commandInfoTestBulk("# Server\r\n"), wantErr: true},
		{name: "retry without version", retry: true, wantErr: true},
		{name: "info denied", infoReply: "-NOPERM INFO denied\r\n", wantErr: true},
		{name: "modules denied", moduleReply: "-NOPERM MODULE denied\r\n", wantErr: true},
		{name: "malformed module", moduleReply: commandInfoTestArray(commandInfoTestBulk("invalid")), wantErr: true},
		{name: "missing module version", moduleReply: commandInfoTestArray(commandInfoTestMap(commandInfoTestBulk("name"), commandInfoTestBulk("custom"))), wantErr: true},
		{name: "command denied", commandReply: "-NOPERM COMMAND denied\r\n", wantErr: true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			if tc.version == "" {
				tc.version = "5.0.14"
			}
			ln, err := net.Listen("tcp", "127.0.0.1:0")
			if err != nil {
				t.Fatal(err)
			}
			t.Cleanup(func() { _ = ln.Close() })
			var connections atomic.Int32
			go func() {
				for {
					conn, err := ln.Accept()
					if err != nil {
						return
					}
					id := connections.Add(1)
					go serveTestRESPConn(conn, func(command string) string {
						switch command {
						case "hello":
							return "-ERR unknown command 'hello'\r\n"
						case "info":
							if tc.retry && id >= 3 {
								return commandInfoTestBulk("# Clients\r\nconnected_clients:1\r\n")
							}
							if tc.infoReply != "" {
								return tc.infoReply
							}
							return commandInfoTestBulk("# Server\r\nredis_version:" + tc.version + "\r\n")
						case "module":
							if tc.retry && id == 2 {
								_ = conn.Close()
								return ""
							}
							if tc.moduleReply != "" {
								return tc.moduleReply
							}
							return commandInfoTestArray(commandInfoTestMap(
								commandInfoTestBulk("name"), commandInfoTestBulk("custom"),
								commandInfoTestBulk("ver"), commandInfoTestInt(int64(id)),
							))
						case "command":
							if tc.commandReply != "" {
								return tc.commandReply
							}
							return commandInfoTestArray(commandInfoTestEntry6(fmt.Sprintf("custom.read%d", id)))
						default:
							return "+OK\r\n"
						}
					})
				}
			}()
			maxRetries := -1
			if tc.retry {
				maxRetries = 1
			}
			client := NewClient(&Options{
				Addr: ln.Addr().String(), Protocol: 2, DisableIdentity: true, MaxRetries: maxRetries,
				// Force fallback to borrow a different connection. Its identity and
				// COMMAND reply must still describe the same server.
				ConnMaxLifetime: time.Nanosecond,
			})
			t.Cleanup(func() { _ = client.Close() })
			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
			defer cancel()
			metadata, err := client.baseClient.fetchCommandMetadata(ctx)
			if tc.wantErr {
				if err == nil {
					t.Fatal("unverifiable metadata was accepted")
				}
				if tc.retry && (connections.Load() < 3 || !strings.Contains(err.Error(), "no server version")) {
					t.Fatalf("retry did not reject the missing version: connections=%d, err=%v", connections.Load(), err)
				}
				return
			}
			if err != nil {
				t.Fatal(err)
			}
			if metadata.serverName != "" || metadata.serverVersion != tc.version || metadata.serverFingerprint != tc.version+"|custom:2" ||
				metadata.records["custom.read2"] == nil || metadata.records["custom.read1"] != nil {
				t.Fatalf("metadata came from mismatched connections: %+v", metadata)
			}
			view := buildCommandMetadataViewForServerWithLegacy(metadata.records, nil, metadata.serverName, metadata.serverVersion, metadata.legacyRecords)
			if !view.routingTable["custom.read2"].valid || isCacheableInView(view, makeCmd("custom.read2", "key")) {
				t.Fatal("module metadata without server identification must enable routing but remain uncached")
			}
		})
	}
}

func TestCommandMetadataReconnectWithoutHello(t *testing.T) {
	var phase, fetches atomic.Int32
	client := NewClient(&Options{
		Protocol: 2, DisableIdentity: true, MaxRetries: -1, PoolSize: 1,
		CommandMetadata: &CommandMetadataConfig{Mode: CommandMetadataPreferLive},
		Dialer: func(context.Context, string, string) (net.Conn, error) {
			client, server := net.Pipe()
			go serveTestRESPConn(server, func(command string) string {
				switch command {
				case "hello":
					return "-ERR unknown command 'hello'\r\n"
				case "info":
					return commandInfoTestBulk("# Server\r\nredis_version:5.0.14\r\n")
				case "module":
					return commandInfoTestArray(commandInfoTestMap(
						commandInfoTestBulk("name"), commandInfoTestBulk("custom"),
						commandInfoTestBulk("ver"), commandInfoTestInt(int64(phase.Load())),
					))
				case "command":
					fetches.Add(1)
					return commandInfoTestArray(commandInfoTestEntry6(fmt.Sprintf("custom.read%d", phase.Load())))
				case "ping":
					return "+PONG\r\n"
				default:
					return "+OK\r\n"
				}
			})
			return client, nil
		},
	})
	t.Cleanup(func() { _ = client.Close() })
	ctx := context.Background()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Fatal(err)
	}
	if !waitForCondition(t, time.Second, func() bool { return client.metadataView().live }) {
		t.Fatal("initial metadata did not load")
	}
	for _, moduleVersion := range []int32{0, 1} {
		command := fmt.Sprintf("custom.read%d", moduleVersion)
		previous := client.metadataView()
		before := fetches.Load()
		phase.Store(moduleVersion)
		if err := client.connPool.(*pool.ConnPool).Filter(func(*pool.Conn) bool { return true }); err != nil {
			t.Fatal(err)
		}
		if err := client.Ping(ctx).Err(); err != nil {
			t.Fatal(err)
		}
		if !waitForCondition(t, 3*time.Second, func() bool {
			view := client.metadataView()
			return fetches.Load() > before && view.live && view.records[command] != nil &&
				client.cmdMeta.serverFingerprint() == fmt.Sprintf("5.0.14|custom:%d", moduleVersion)
		}) {
			t.Fatal("reconnect did not verify the server identity")
		}
		waitForCommandMetadataRefresh(t, client.cmdMeta)
		view := client.metadataView()
		if moduleVersion == 0 && view != previous {
			t.Fatal("unchanged identity rebuilt the metadata view")
		}
	}
}

func TestCommandMetadataServerChangeRefreshesLiveView(t *testing.T) {
	var phase atomic.Int32
	var calls atomic.Int32
	s := newCommandMetadataStore(&CommandMetadataConfig{Mode: CommandMetadataPreferLive},
		func(context.Context) (commandMetadataFetchResult, error) {
			calls.Add(1)
			if phase.Load() == 0 {
				return testCommandMetadataFetchResultFor(
					map[string]*CommandInfo{"srva.get": testReadOnlyCommandInfo("srva.get")},
					"8.10.0", "8.10.0|srvA",
				), nil
			}
			return testCommandMetadataFetchResultFor(
				map[string]*CommandInfo{"srvb.get": testReadOnlyCommandInfo("srvb.get")},
				"8.11.0", "8.11.0|srvB",
			), nil
		})
	defer s.stopAndJoin()

	s.onServerHello("8.10.0|srvA")
	if !waitForCondition(t, 5*time.Second, func() bool {
		return s.view().live && isCacheableInView(s.view(), makeCmd("srva.get", "k"))
	}) {
		t.Fatal("first server's live view never published")
	}
	// The same identity must not refetch.
	settled := calls.Load()
	s.onServerHello("8.10.0|srvA")
	time.Sleep(30 * time.Millisecond)
	if calls.Load() != settled {
		t.Errorf("unchanged server identity refetched: %d -> %d", settled, calls.Load())
	}
	// A changed identity must retire and replace the live view.
	phase.Store(1)
	s.onServerHello("8.11.0|srvB")
	if !waitForCondition(t, 5*time.Second, func() bool {
		v := s.view()
		return v.live && isCacheableInView(v, makeCmd("srvb.get", "k")) &&
			!isCacheableInView(v, makeCmd("srva.get", "k"))
	}) {
		t.Fatal("server change did not refresh the live view")
	}
}

func TestCommandMetadataServerUpgradeEnablesLiveOnlyCSC(t *testing.T) {
	keyed := testReadOnlyCommandInfo("myext.get")
	var upgraded atomic.Bool
	var calls atomic.Int32
	s := newCommandMetadataStore(&CommandMetadataConfig{Mode: CommandMetadataPreferLive},
		func(context.Context) (commandMetadataFetchResult, error) {
			calls.Add(1)
			version := "7.4.0"
			if upgraded.Load() {
				version = "8.10.0"
			}
			return testCommandMetadataFetchResultFor(
				map[string]*CommandInfo{"myext.get": keyed}, version, version,
			), nil
		})
	defer s.stopAndJoin()

	s.onServerHello("7.4.0")
	if !waitForCondition(t, 5*time.Second, func() bool {
		return s.view().live && !isCacheableInView(s.view(), makeCmd("myext.get", "k"))
	}) {
		t.Fatal("pre-8.10 live view was not published fail-closed")
	}
	// Connection churn must not refetch the same identity.
	settled := calls.Load()
	s.onConnInit()
	time.Sleep(30 * time.Millisecond)
	if calls.Load() != settled {
		t.Errorf("live store refetched without a server change: %d -> %d", settled, calls.Load())
	}
	meta, ok := routingLookupMeta(s.view(), makeCmd("myext.get", "k"))
	if !ok || !meta.readOnly {
		t.Fatal("pre-8.10 record was lost to the routing consumer")
	}
	// An upgrade can enable a live-only command after refresh.
	upgraded.Store(true)
	s.onServerHello("8.10.0")
	if !waitForCondition(t, 5*time.Second, func() bool {
		return s.view().live && isCacheableInView(s.view(), makeCmd("myext.get", "k"))
	}) {
		t.Fatal("upgrade did not enable the live-only CSC record")
	}
}

func TestCommandMetadataStraddledFetchNotPublished(t *testing.T) {
	// An old-server fetch must not publish after an identity change.
	keyed := testReadOnlyCommandInfo("old.get")
	started := make(chan struct{}, 4)
	release := make(chan struct{})
	var phase atomic.Int32
	s := newCommandMetadataStore(&CommandMetadataConfig{Mode: CommandMetadataPreferLive},
		func(context.Context) (commandMetadataFetchResult, error) {
			if phase.Load() == 0 {
				started <- struct{}{}
				<-release
				return testCommandMetadataFetchResultFor(
					map[string]*CommandInfo{"old.get": keyed},
					"8.10.0", "srvOld",
				), nil
			}
			return testCommandMetadataFetchResultFor(
				nil, "8.10.0", "srvNew",
			), nil
		})
	defer s.stopAndJoin()

	s.onServerHello("srvOld")
	select {
	case <-started:
	case <-time.After(2 * time.Second):
		t.Fatal("first fetch never started")
	}
	phase.Store(1)
	s.onServerHello("srvNew") // identity changes while fetch #1 is in flight
	close(release)            // fetch #1 completes with the OLD server's data

	if !waitForCondition(t, 5*time.Second, func() bool { return s.view().live }) {
		t.Fatal("new server's view never published")
	}
	if isCacheableInView(s.view(), makeCmd("old.get", "k")) {
		t.Fatal("straddled fetch published the old server's metadata")
	}
}

func TestCSCServerCompatibilityKeepsSnapshotNegatives(t *testing.T) {
	// Compatibility exclusions belong in the shared record, not only the CSC table.
	live := map[string]*CommandInfo{
		"ts.info":     testReadOnlyCommandInfo("ts.info"),
		"eval_ro":     testReadOnlyCommandInfo("eval_ro"),
		"ttl":         testReadOnlyCommandInfo("ttl"),
		"fcall_ro":    testReadOnlyCommandInfo("fcall_ro"),
		"custom.read": testReadOnlyCommandInfo("custom.read"),
		"get":         testReadOnlyCommandInfo("get"),
	}
	redis810 := buildCommandMetadataViewForServer(live, nil, "redis", "8.10.0")
	if !isCacheableInView(redis810, makeCmd("ts.info", "k")) {
		t.Error("Redis 8.10+ live record did not remain authoritative")
	}
	for _, tc := range []struct{ name, server, version string }{
		{"older Redis", "redis", "7.4.0"},
		{"other server", "other", "8.10.0"},
		{"unknown server", "", "8.10.0"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			view := buildCommandMetadataViewForServer(live, nil, tc.server, tc.version)
			if !isCacheableInView(view, makeCmd("get", "k")) {
				t.Error("known GET must remain cacheable")
			}
			for _, name := range []string{"eval_ro", "fcall_ro", "custom.read"} {
				if isCacheableInView(view, makeCmd(name, "k")) {
					t.Errorf("%s must remain uncached without complete CSC signals", name)
				}
			}
			if isCacheableInView(view, makeCmd("ts.info", "k")) {
				t.Error("conservative live record cleared the snapshot's dont_cache signal")
			}
			if !commandRecordHas(view.records["ts.info"], "dont_cache", true) {
				t.Error("conservative compatibility correction was not written to the shared record")
			}
			if !commandRecordHas(view.records["eval_ro"], "script_runner", false) {
				t.Error("conservative script_runner correction was not written to the shared record")
			}
			if !commandRecordHas(view.records["ttl"], "nondeterministic_output", true) {
				t.Error("conservative correction did not retain the snapshot's nondeterministic signal")
			}
			if isCacheableInView(view, makeCmd("ttl", "k")) {
				t.Error("conservative live TTL record lost its snapshot exclusion")
			}
			if got, want := view.cscTable["ts.info"], cscDeriveMeta(view.records["ts.info"]); got != want {
				t.Fatalf("CSC metadata was not derived solely from the resolved record: got %+v, want %+v", got, want)
			}
			if !commandRecordHas(commandInfoSnapshotByName()["ts.info"], "dont_cache", true) {
				t.Fatal("test premise: snapshot ts.info must carry dont_cache")
			}
			// Application overrides may replace the correction.
			view = buildCommandMetadataViewForServer(live, map[string]*CommandInfo{"ts.info": live["ts.info"]}, tc.server, tc.version)
			if !isCacheableInView(view, makeCmd("ts.info", "k")) {
				t.Error("an explicit application override must not be clamped")
			}
		})
	}
}

func TestCSCDeriveMetaRejectsOverflowingKeynum(t *testing.T) {
	// Overflow must not wrap onto a valid position.
	info := &CommandInfo{
		Name:  "evil",
		Flags: []string{"readonly"},
		KeySpecs: []KeySpec{{
			BeginSearch: "index", Index: 32760, FindKeys: "keynum",
			KeyNumIdx: 10, FirstKey: 11, KeyStep: 1,
		}},
	}
	if m := cscDeriveMeta(info); m.extract != cscKeyExtractNone {
		t.Errorf("overflowing keynum positions must derive no extraction, got %+v", m)
	}

	// Invalid offsets must not cancel into a plausible position.
	info.KeySpecs[0] = KeySpec{
		BeginSearch: "index", Index: math.MaxInt, FindKeys: "keynum",
		KeyNumIdx: 1 - math.MaxInt, FirstKey: 2 - math.MaxInt, KeyStep: 1,
	}
	if m := cscDeriveMeta(info); m.extract != cscKeyExtractNone {
		t.Errorf("canceling malformed keynum positions must derive no extraction, got %+v", m)
	}
}

func TestCommandsInfoMalformedKeyPositionsFailClosed(t *testing.T) {
	// firstkey 257 must not wrap into int8 position 1.
	raw := "*2\r\n" +
		"*6\r\n$3\r\nbad\r\n:-1\r\n*1\r\n$8\r\nreadonly\r\n:257\r\n:257\r\n:1\r\n" +
		"*6\r\n$4\r\ngood\r\n:2\r\n*1\r\n$8\r\nreadonly\r\n:1\r\n:1\r\n:1\r\n"
	cmd := NewCommandsInfoCmd(context.Background(), "command")
	if err := cmd.readReply(proto.NewReader(strings.NewReader(raw))); err != nil {
		t.Fatal(err)
	}
	bad, exists := cmd.Val()["bad"]
	if !exists || bad != nil {
		t.Errorf("out-of-range key positions must tombstone the record, got %+v (exists=%v)", bad, exists)
	}
	if good := cmd.Val()["good"]; good == nil || good.FirstKeyPos != 1 || good.StepCount != 1 {
		t.Errorf("in-range key positions must parse, got %+v", good)
	}

	badArity := "*2\r\n" +
		"*6\r\n$3\r\nbad\r\n:128\r\n*1\r\n$8\r\nreadonly\r\n:1\r\n:1\r\n:1\r\n" +
		"*6\r\n$4\r\ngood\r\n:2\r\n*1\r\n$8\r\nreadonly\r\n:1\r\n:1\r\n:1\r\n"
	cmd = NewCommandsInfoCmd(context.Background(), "command")
	if err := cmd.readReply(proto.NewReader(strings.NewReader(badArity))); err != nil {
		t.Fatalf("out-of-range command arity aborted the reply: %v", err)
	}
	if bad, exists := cmd.Val()["bad"]; !exists || bad != nil {
		t.Fatalf("out-of-range command arity must tombstone the record, got %+v (exists=%v)", bad, exists)
	}
	if cmd.Val()["good"] == nil {
		t.Fatal("out-of-range command arity discarded the following valid record")
	}

	overflowKeySpec := "*2\r\n$4\r\nspec\r\n" +
		"*2\r\n$5\r\nindex\r\n:2147483648\r\n"
	if ok, err := readKeySpecSectionChecked(
		proto.NewReader(strings.NewReader(overflowKeySpec)), &KeySpec{}, true,
	); err != nil || ok {
		t.Fatal("out-of-range key-spec position must fail instead of narrowing")
	}
}

func TestCommandsInfoRejectsExcessiveSubcommandDepth(t *testing.T) {
	entry := "*10\r\n$3\r\ncmd\r\n:-1\r\n*0\r\n:0\r\n:0\r\n:0\r\n" +
		"*0\r\n*0\r\n*0\r\n*0\r\n"
	for range maxCommandInfoDepth + 1 {
		entry = "*10\r\n$3\r\ncmd\r\n:-1\r\n*0\r\n:0\r\n:0\r\n:0\r\n" +
			"*0\r\n*0\r\n*0\r\n*1\r\n" + entry
	}

	cmd := NewCommandsInfoCmd(context.Background(), "command")
	err := cmd.readReply(proto.NewReader(strings.NewReader("*1\r\n" + entry)))
	if err == nil || !strings.Contains(err.Error(), "maximum depth") {
		t.Fatalf("deeply nested subcommands returned %v, want maximum-depth error", err)
	}
}

func TestHelloServerFingerprint(t *testing.T) {
	hello := map[string]interface{}{
		"server":  "redis",
		"version": "8.10.0",
		"modules": []interface{}{
			map[interface{}]interface{}{"name": "timeseries", "ver": int64(81000)},
			map[string]interface{}{"name": "bf", "ver": int64(81000)},
			[]interface{}{"name", "json", "ver", int64(81000)}, // RESP2
		},
	}
	fp := helloServerFingerprint(hello)
	if fp != "redis:8.10.0|bf:81000|json:81000|timeseries:81000" {
		t.Errorf("fingerprint = %q", fp)
	}
	for _, server := range []interface{}{"other", nil, 42} {
		hello["server"] = server
		if helloServerFingerprint(hello) == fp {
			t.Errorf("server %v kept Redis's fingerprint", server)
		}
	}
	if helloServerFingerprint(map[string]interface{}{}) != "" {
		t.Error("empty reply must produce an empty fingerprint")
	}
}

func TestCommandMetadataRetryCapStopsSelfRetry(t *testing.T) {
	oldMin, oldCap := cmdMetaBackoffMin, cmdMetaRetryCap
	cmdMetaBackoffMin, cmdMetaRetryCap = time.Millisecond, 3
	defer func() { cmdMetaBackoffMin, cmdMetaRetryCap = oldMin, oldCap }()

	var calls atomic.Int32
	s := newCommandMetadataStore(&CommandMetadataConfig{Mode: CommandMetadataPreferLive},
		func(context.Context) (commandMetadataFetchResult, error) {
			calls.Add(1)
			return commandMetadataFetchResult{}, errors.New("NOPERM this user has no permissions to run the 'command' command")
		})
	defer s.stopAndJoin()

	s.onConnInit()
	if !waitForCondition(t, 5*time.Second, func() bool { return calls.Load() >= 3 }) {
		t.Fatal("retries never ran")
	}
	settled := calls.Load()
	time.Sleep(50 * time.Millisecond)
	if calls.Load() != settled {
		t.Errorf("worker kept self-retrying past the cap: %d -> %d", settled, calls.Load())
	}
	// An external trigger permits one fresh attempt.
	s.onConnInit()
	if !waitForCondition(t, 2*time.Second, func() bool { return calls.Load() == settled+1 }) {
		t.Errorf("external trigger after the cap did not attempt: %d -> %d", settled, calls.Load())
	}
}

func TestCommandMetadataViewChangeCancelsFulfill(t *testing.T) {
	for _, coalesced := range []bool{false, true} {
		t.Run(fmt.Sprintf("coalesced=%v", coalesced), func(t *testing.T) {
			cache := NewLocalCache(CacheConfig{MaxEntries: 16})
			s := newCommandMetadataStoreForLive(nil, nil)
			c := &baseClient{opt: &Options{Protocol: 3}, csc: cache, cmdMeta: s}
			view := c.metadataView()
			for _, changed := range []bool{true, false} {
				key := "get:k"
				token, fetch := cache.Reserve(key, []string{"k"})
				if !fetch {
					t.Fatal("Reserve should fetch")
				}
				var overrides map[string]*CommandInfo
				if changed {
					overrides = map[string]*CommandInfo{"get": nil}
				}
				// A fresh pointer with the same decisions still permits publication.
				s.current.Store(buildCommandMetadataView(nil, overrides))
				raw := []byte("$1\r\nv\r\n")
				if coalesced {
					cmd := NewStringCmd(context.Background(), "get", "k")
					req := &cscMissReq{cmd: cmd, cacheKey: key, token: token, view: view, done: make(chan error, 1)}
					mc := &cscMissCoalescer{c: c}
					mc.applyAndSettle(req, raw, 0, 0)
					if err := <-req.done; err != nil || cmd.Val() != "v" {
						t.Fatalf("fetched reply = %q, %v", cmd.Val(), err)
					}
				} else {
					c.fulfillCached(key, token, &cscFetchCapture{raw: raw}, view)
				}
				ctx, cancel := context.WithTimeout(context.Background(), time.Second)
				_, cached := cache.Get(ctx, key)
				cancel()
				if cached == changed {
					t.Fatalf("changed=%v: cached=%v, want %v", changed, cached, !changed)
				}
			}
		})
	}
}

func TestCommandMetadataRefreshSkipsRetiredEntries(t *testing.T) {
	cache := NewLocalCache(CacheConfig{MaxEntries: 16})
	s := newCommandMetadataStore(&CommandMetadataConfig{Overrides: map[string]*CommandInfo{"get": nil}}, nil)
	pooler := &erroringPooler{}
	c := &baseClient{opt: &Options{}, csc: cache, cmdMeta: s, connPool: pooler, cscKeyPrefix: "p:"}
	rawKey, _ := buildCacheKey(makeCmd("get", "k"))
	key := cscEntryKey(c.cscKeyPrefix, defaultCommandMetadataView().cscFingerprint, rawKey)
	n, err := c.refreshInvalidatedBatch(context.Background(), []cscRefreshTarget{{cacheKey: key, redisKeys: []string{"p:k"}}})
	if err != nil || n != 0 || pooler.gets.Load() != 0 || cache.Len() != 0 {
		t.Fatalf("retired refresh: published=%d err=%v pool gets=%d entries=%d", n, err, pooler.gets.Load(), cache.Len())
	}
}

// TestCommandMetadataPreferLiveE2E runs the dynamic path against Redis.
func TestCommandMetadataPreferLiveE2E(t *testing.T) {
	if testing.Short() {
		t.Skip("requires a running Redis server")
	}
	addr := "localhost:6379"
	if v := os.Getenv("REDIS_ADDR"); v != "" {
		addr = v
	} else if p := os.Getenv("REDIS_PORT"); p != "" {
		addr = "localhost:" + p
	}

	cache := NewLocalCache(CacheConfig{MaxEntries: 32})
	client := NewClient(&Options{
		Addr:            addr,
		Protocol:        3,
		ClientSideCache: cache,
		CommandMetadata: &CommandMetadataConfig{Mode: CommandMetadataPreferLive},
		PoolSize:        1,
		MaxRetries:      -1,
	})
	defer client.Close()
	ctx := context.Background()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no server at %s: %v", addr, err)
	}
	identity, err := client.baseClient.fetchMetadataIdentity(ctx, nil)
	if err != nil {
		t.Fatal(err)
	}
	if !commandMetadataSupportsCSC(identity.serverName, identity.serverVersion) {
		t.Skipf("live command metadata requires Redis 8.10 or newer (server is %s %s)", identity.serverName, identity.serverVersion)
	}

	if client.baseClient.cmdMeta == nil {
		t.Fatal("PreferLive client must carry a metadata store")
	}
	if !waitForCondition(t, 10*time.Second, func() bool { return client.baseClient.metadataView().live }) {
		t.Fatal("live view never published against a real 8.10 server")
	}
	live := client.baseClient.metadataView()

	// The live view must reproduce normative decisions.
	for cmd, want := range map[Cmder]bool{
		makeCmd("get", "k"):             true,
		makeCmd("mget", "a", "b"):       true,
		makeCmd("touch", "k"):           false,
		makeCmd("json.mget", "a", "$"):  false,
		makeCmd("memory", "usage", "k"): false,
		makeCmd("blpop", "k", "0"):      false,
	} {
		if got := isCacheableInView(live, cmd); got != want {
			t.Errorf("live view: isCacheable(%v) = %v, want %v", cmd.Args(), got, want)
		}
	}

	// Caching must work with the live fingerprint.
	mutator := NewClient(&Options{Addr: addr})
	defer mutator.Close()
	key := "cmdmeta:e2e:k"
	if err := mutator.Set(ctx, key, "v1", 0).Err(); err != nil {
		t.Fatal(err)
	}
	defer mutator.Del(ctx, key)

	deadline := time.Now().Add(3 * time.Second)
	for cache.Len() < 1 && time.Now().Before(deadline) {
		if err := client.Get(ctx, key).Err(); err != nil {
			t.Fatal(err)
		}
		time.Sleep(20 * time.Millisecond)
	}
	if cache.Len() < 1 {
		t.Fatal("entry never cached under the live view")
	}

	// An unchanged metadata fetch must preserve already cached replies.
	entries := cache.Len()
	if _, err := client.baseClient.fetchCommandMetadata(ctx); err != nil {
		t.Fatal(err)
	}
	if cache.Len() != entries {
		t.Fatal("unchanged metadata fetch evicted cached entries")
	}

	if err := mutator.Set(ctx, key, "v2", 0).Err(); err != nil {
		t.Fatal(err)
	}
	fresh := waitForCondition(t, 5*time.Second, func() bool {
		v, err := client.Get(ctx, key).Result()
		return err == nil && v == "v2"
	})
	if !fresh {
		t.Fatal("invalidation did not reach the fingerprinted entry")
	}
}