fix: parse CLUSTER NODES endpoints IPv6-aware in stale-address detection

CLUSTER NODES renders IPv6 endpoints bracketed ([fd00::2]:6379@16379).
Slicing the endpoint at its last colon kept the brackets, so a failing
IPv6 peer at its current address never compared equal to the bare pod
IP Kubernetes reports - misclassifying it as moved and issuing spurious
CLUSTER MEETs plus requeues on IPv6 clusters.

Extract the host with net.SplitHostPort (after trimming the @cport and
optional ,hostname suffix), which unbrackets IPv6 and leaves IPv4
untouched.

Signed-off-by: Matan David <matan.david@eon.io>

Author: matka12

internal/valkey/clusterstate.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"net"
 	"slices"
 	"strconv"
 	"strings"
@@ -396,6 +397,27 @@ func (s *ClusterState) FindNodeById(id string) *NodeState {
 	return nil
 }
 
+// hostFromClusterNodesEndpoint extracts the bare host from a CLUSTER NODES
+// endpoint field (<ip:port@cport[,hostname]>). IPv6 hosts appear bracketed
+// ([fd00::2]:6379@16379); net.SplitHostPort unbrackets them so the result
+// compares equal to the bare pod IP Kubernetes reports. Returns "" when the
+// field has no parsable host:port part.
+func hostFromClusterNodesEndpoint(endpoint string) string {
+	// Drop the cluster-bus suffix and the optional ,hostname after it.
+	if i := strings.Index(endpoint, "@"); i != -1 {
+		endpoint = endpoint[:i]
+	}
+	if host, _, err := net.SplitHostPort(endpoint); err == nil {
+		return host
+	}
+	// Fallback for entries with no port (shouldn't occur in CLUSTER NODES,
+	// but keep the previous last-colon behavior rather than dropping them).
+	if i := strings.LastIndex(endpoint, ":"); i != -1 {
+		return strings.Trim(endpoint[:i], "[]")
+	}
+	return ""
+}
+
 // StaleAddressPeer pairs a viewer node with a live cluster member whose
 // address in the viewer's node table is outdated.
 type StaleAddressPeer struct {
@@ -442,11 +464,7 @@ func (s *ClusterState) FindStaleAddressPeers() []StaleAddressPeer {
 			if !ok {
 				continue
 			}
-			idx := strings.LastIndex(fields[1], ":")
-			if idx == -1 {
-				continue
-			}
-			if address := fields[1][:idx]; address != peer.Address {
+			if address := hostFromClusterNodesEndpoint(fields[1]); address != "" && address != peer.Address {
 				stale = append(stale, StaleAddressPeer{Viewer: viewer, Live: peer})
 			}
 		}

internal/valkey/clusterstate_test.go

@@ -640,3 +640,62 @@ func TestClusterState_FindNodeById(t *testing.T) {
 		t.Error("expected nil for unknown id")
 	}
 }
+
+func TestHostFromClusterNodesEndpoint(t *testing.T) {
+	cases := []struct{ in, want string }{
+		{"10.0.1.2:6379@16379", "10.0.1.2"},
+		{"10.0.1.2:6379@16379,node-1.example", "10.0.1.2"},
+		{"[fd00::2]:6379@16379", "fd00::2"},
+		{"[fd00::2]:6379@16379,node-1.example", "fd00::2"},
+		{"[fd00::2]:6379", "fd00::2"},
+		{"10.0.1.2:6379", "10.0.1.2"},
+		{"garbage", ""},
+	}
+	for _, c := range cases {
+		if got := hostFromClusterNodesEndpoint(c.in); got != c.want {
+			t.Errorf("hostFromClusterNodesEndpoint(%q) = %q, want %q", c.in, got, c.want)
+		}
+	}
+}
+
+func TestClusterState_FindStaleAddressPeers_IPv6(t *testing.T) {
+	node := func(id, address, clusterNodes string) *NodeState {
+		return &NodeState{Id: id, Address: address, ClusterNodes: clusterNodes}
+	}
+
+	t.Run("failing IPv6 peer at its current address is not stale", func(t *testing.T) {
+		// Bracketed table entry vs bare pod IP must compare equal.
+		a := node("aaa", "fd00::1",
+			"aaa [fd00::1]:6379@16379 myself,master - 0 0 1 connected 0-8191\n"+
+				"bbb [fd00::2]:6379@16379 master,fail? - 0 0 2 connected 8192-16383\n")
+		b := node("bbb", "fd00::2",
+			"bbb [fd00::2]:6379@16379 myself,master - 0 0 2 connected 8192-16383\n"+
+				"aaa [fd00::1]:6379@16379 master - 0 0 1 connected 0-8191\n")
+		state := &ClusterState{Shards: []*ShardState{
+			{Id: "s1", PrimaryId: "aaa", Nodes: []*NodeState{a}},
+			{Id: "s2", PrimaryId: "bbb", Nodes: []*NodeState{b}},
+		}}
+
+		if stale := state.FindStaleAddressPeers(); len(stale) != 0 {
+			t.Fatalf("expected no stale pairs for current bracketed IPv6 address, got %d", len(stale))
+		}
+	})
+
+	t.Run("moved IPv6 peer is stale", func(t *testing.T) {
+		a := node("aaa", "fd00::11",
+			"aaa [fd00::11]:6379@16379 myself,master - 0 0 1 connected 0-8191\n"+
+				"bbb [fd00::2]:6379@16379 master,fail? - 0 0 2 connected 8192-16383\n")
+		b := node("bbb", "fd00::12",
+			"bbb [fd00::12]:6379@16379 myself,master - 0 0 2 connected 8192-16383\n"+
+				"aaa [fd00::11]:6379@16379 master - 0 0 1 connected 0-8191\n")
+		state := &ClusterState{Shards: []*ShardState{
+			{Id: "s1", PrimaryId: "aaa", Nodes: []*NodeState{a}},
+			{Id: "s2", PrimaryId: "bbb", Nodes: []*NodeState{b}},
+		}}
+
+		stale := state.FindStaleAddressPeers()
+		if len(stale) != 1 || stale[0].Viewer.Id != "aaa" || stale[0].Live.Id != "bbb" {
+			t.Fatalf("expected exactly aaa->bbb, got %v", stale)
+		}
+	})
+}