Address Redis Cluster mode review follow-ups for #5153 (#5210)

Address Redis Cluster mode review follow-ups

Three small follow-ups from the PR #5153 review that were deferred so the
core cluster-mode support could land:

- Document the cluster slot invariant for storeUpstreamTokensScript. The
  script reads oldUserID from KEYS[1] inside its body to keep the user-set
  bookkeeping atomic, so the user-set keys are constructed dynamically from
  ARGV[4] rather than declared as KEYS. The new comment makes the
  {ns:name} hash-tag requirement explicit, so a future refactor that
  rebuilds the prefix without the tag will be caught in review rather than
  silently regressing on standalone Redis and exploding with CROSSSLOT on
  a real cluster.

- Filter SMEMBERS results in GetAllUpstreamTokens, DeleteUpstreamTokens,
  and GetLatestUpstreamTokensForUser to entries that share the storage
  instance's keyPrefix, warn-logging anything dropped. A stray un-prefixed
  member (legacy data, an external admin op, a test fixture) would today
  surface as CROSSSLOT under cluster while passing on standalone; this
  turns it into a logged warning. The defensive filter is a pure helper
  with table-driven tests, plus a behavior test that captures slog output
  to prove the wiring at all three call sites.

- Align the operator CRD `tls` doc with the storage-layer comment so that
  crd-api.md and `kubectl explain` both describe TLS as applying to
  "Redis/Valkey master or cluster nodes" — previously cluster-mode users
  reading the CRD field would not realise this same field configured TLS
  for cluster-node connections. Regenerated CRD YAML, Helm templates, and
  crd-api.md to match.

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: reyortiz3

cmd/thv-operator/api/v1beta1/mcpexternalauthconfig_types.go

@@ -869,7 +869,7 @@ type RedisStorageConfig struct {
 	// +optional
 	WriteTimeout string `json:"writeTimeout,omitempty"`
 
-	// TLS configures TLS for connections to the Redis/Valkey master.
+	// TLS configures TLS for connections to the Redis/Valkey master or cluster nodes.
 	// Presence of this field enables TLS. Omit to use plaintext.
 	// +optional
 	TLS *RedisTLSConfig `json:"tls,omitempty"`

deploy/charts/operator-crds/files/crds/toolhive.stacklok.dev_mcpexternalauthconfigs.yaml

@@ -489,7 +489,7 @@ spec:
                             type: object
                           tls:
                             description: |-
-                              TLS configures TLS for connections to the Redis/Valkey master.
+                              TLS configures TLS for connections to the Redis/Valkey master or cluster nodes.
                               Presence of this field enables TLS. Omit to use plaintext.
                             properties:
                               caCertSecretRef:
@@ -1798,7 +1798,7 @@ spec:
                             type: object
                           tls:
                             description: |-
-                              TLS configures TLS for connections to the Redis/Valkey master.
+                              TLS configures TLS for connections to the Redis/Valkey master or cluster nodes.
                               Presence of this field enables TLS. Omit to use plaintext.
                             properties:
                               caCertSecretRef:

deploy/charts/operator-crds/files/crds/toolhive.stacklok.dev_virtualmcpservers.yaml

@@ -362,7 +362,7 @@ spec:
                             type: object
                           tls:
                             description: |-
-                              TLS configures TLS for connections to the Redis/Valkey master.
+                              TLS configures TLS for connections to the Redis/Valkey master or cluster nodes.
                               Presence of this field enables TLS. Omit to use plaintext.
                             properties:
                               caCertSecretRef:
@@ -3316,7 +3316,7 @@ spec:
                             type: object
                           tls:
                             description: |-
-                              TLS configures TLS for connections to the Redis/Valkey master.
+                              TLS configures TLS for connections to the Redis/Valkey master or cluster nodes.
                               Presence of this field enables TLS. Omit to use plaintext.
                             properties:
                               caCertSecretRef:

deploy/charts/operator-crds/templates/toolhive.stacklok.dev_mcpexternalauthconfigs.yaml

@@ -492,7 +492,7 @@ spec:
                             type: object
                           tls:
                             description: |-
-                              TLS configures TLS for connections to the Redis/Valkey master.
+                              TLS configures TLS for connections to the Redis/Valkey master or cluster nodes.
                               Presence of this field enables TLS. Omit to use plaintext.
                             properties:
                               caCertSecretRef:
@@ -1801,7 +1801,7 @@ spec:
                             type: object
                           tls:
                             description: |-
-                              TLS configures TLS for connections to the Redis/Valkey master.
+                              TLS configures TLS for connections to the Redis/Valkey master or cluster nodes.
                               Presence of this field enables TLS. Omit to use plaintext.
                             properties:
                               caCertSecretRef:

deploy/charts/operator-crds/templates/toolhive.stacklok.dev_virtualmcpservers.yaml

@@ -365,7 +365,7 @@ spec:
                             type: object
                           tls:
                             description: |-
-                              TLS configures TLS for connections to the Redis/Valkey master.
+                              TLS configures TLS for connections to the Redis/Valkey master or cluster nodes.
                               Presence of this field enables TLS. Omit to use plaintext.
                             properties:
                               caCertSecretRef:
@@ -3319,7 +3319,7 @@ spec:
                             type: object
                           tls:
                             description: |-
-                              TLS configures TLS for connections to the Redis/Valkey master.
+                              TLS configures TLS for connections to the Redis/Valkey master or cluster nodes.
                               Presence of this field enables TLS. Omit to use plaintext.
                             properties:
                               caCertSecretRef:

docs/operator/crd-api.md

@@ -11,6 +11,7 @@ import (
 	"log/slog"
 	"net/url"
 	"slices"
+	"strings"
 	"time"
 
 	"github.com/ory/fosite"
@@ -52,6 +53,35 @@ func warnOnCleanupErr(err error, operation, key string) {
 	}
 }
 
+// filterIndexMembersByPrefix splits SMEMBERS results into members that share
+// this storage instance's key prefix and those that do not. Multi-key
+// operations like MGet and Del fail with CROSSSLOT under Redis Cluster when
+// any member hashes to a different slot, and a stray un-prefixed entry
+// (legacy data, an external admin op, a test fixture) is the most likely
+// source of such drift. Filtering at read time turns a hard cluster failure
+// into a logged warning, while remaining a no-op on standalone Redis.
+func filterIndexMembersByPrefix(prefix string, members []string) (kept, dropped []string) {
+	for _, m := range members {
+		if strings.HasPrefix(m, prefix) {
+			kept = append(kept, m)
+		} else {
+			dropped = append(dropped, m)
+		}
+	}
+	return kept, dropped
+}
+
+// warnDroppedIndexMembers emits a warning for each foreign member returned by
+// filterIndexMembersByPrefix so operators can identify the source of cluster-
+// incompatible entries.
+func warnDroppedIndexMembers(operation, indexKey, expectedPrefix string, dropped []string) {
+	for _, m := range dropped {
+		slog.Warn("dropping foreign index member to prevent CROSSSLOT",
+			"operation", operation, "indexKey", indexKey,
+			"expectedPrefix", expectedPrefix, "member", m)
+	}
+}
+
 // RedisStorage implements the Storage interface backed by Redis.
 // Supports standalone mode (single endpoint), Sentinel failover mode, and
 // Cluster mode. It provides distributed storage for OAuth2 tokens, authorization
@@ -692,6 +722,17 @@ func (s *storedUpstreamTokens) toUpstreamTokens() *UpstreamTokens {
 // ARGV[2] = TTL in milliseconds
 // ARGV[3] = new UserID ("" if no user)
 // ARGV[4] = user upstream set key prefix (e.g. "thv:auth:{ns:name}:user:upstream:")
+//
+// Cluster slot invariant: this script reads oldUserID from KEYS[1] inside the
+// script body (atomic with the rest of the work), so the user-set keys are
+// constructed dynamically as `ARGV[4] .. userID` rather than being passed as
+// declared KEYS. Every dynamically-built key MUST therefore inherit the
+// `{ns:name}` hash tag that is baked into ARGV[4] (s.keyPrefix). All callers
+// derive ARGV[4] from s.keyPrefix, which DeriveKeyPrefix builds with the
+// `{ns:name}` hash tag, so user-set keys land on the same Redis Cluster slot
+// as KEYS[1] and KEYS[2]. A future refactor that strips the hash tag — or
+// rebuilds setPrefix from raw inputs without the tag — will silently pass on
+// standalone Redis and fail with CROSSSLOT under Cluster.
 var storeUpstreamTokensScript = redis.NewScript(`
 local oldUserID = ""
 local existing = redis.call('GET', KEYS[1])
@@ -899,6 +940,9 @@ func (s *RedisStorage) GetAllUpstreamTokens(ctx context.Context, sessionID strin
 		return nil, fmt.Errorf("failed to get upstream token index: %w", err)
 	}
 
+	providerKeys, dropped := filterIndexMembersByPrefix(s.keyPrefix, providerKeys)
+	warnDroppedIndexMembers("GetAllUpstreamTokens", idxKey, s.keyPrefix, dropped)
+
 	if len(providerKeys) == 0 {
 		return result, nil
 	}
@@ -961,6 +1005,9 @@ func (s *RedisStorage) DeleteUpstreamTokens(ctx context.Context, sessionID strin
 		return fmt.Errorf("failed to get upstream token index: %w", err)
 	}
 
+	providerKeys, dropped := filterIndexMembersByPrefix(s.keyPrefix, providerKeys)
+	warnDroppedIndexMembers("DeleteUpstreamTokens", idxKey, s.keyPrefix, dropped)
+
 	if len(providerKeys) == 0 {
 		return fmt.Errorf("%w: %w", ErrNotFound, fosite.ErrNotFound.WithHint("Upstream tokens not found"))
 	}
@@ -1017,6 +1064,10 @@ func (s *RedisStorage) GetLatestUpstreamTokensForUser(ctx context.Context, userI
 	if err != nil && !errors.Is(err, redis.Nil) {
 		return nil, fmt.Errorf("failed to get user upstream index: %w", err)
 	}
+
+	members, dropped := filterIndexMembersByPrefix(s.keyPrefix, members)
+	warnDroppedIndexMembers("GetLatestUpstreamTokensForUser", setKey, s.keyPrefix, dropped)
+
 	if len(members) == 0 {
 		return nil, fmt.Errorf("%w: %w", ErrNotFound, fosite.ErrNotFound.WithHint("Upstream tokens not found"))
 	}

pkg/authserver/storage/redis.go

@@ -8,9 +8,12 @@
 package storage
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"log/slog"
 	"net/url"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -2723,3 +2726,144 @@ func runDCRConcurrentAccess(
 	assert.Zero(t, atomic.LoadInt32(&storeErrCount), "no concurrent Store should have errored")
 	assert.Zero(t, atomic.LoadInt32(&getErrCount), "no concurrent Get should have errored")
 }
+
+// captureWarnLogs swaps in a buffered slog handler at warn level for the
+// duration of the test and returns the captured output. Process-global, not
+// safe for t.Parallel().
+func captureWarnLogs(t *testing.T) *bytes.Buffer {
+	t.Helper()
+	var buf bytes.Buffer
+	handler := slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelWarn})
+	orig := slog.Default()
+	slog.SetDefault(slog.New(handler))
+	t.Cleanup(func() { slog.SetDefault(orig) })
+	return &buf
+}
+
+// assertForeignDropWarn checks that the warn log emitted for a filtered
+// CROSSSLOT-defending op names the operation, the dropped member, and the
+// expected key prefix.
+func assertForeignDropWarn(t *testing.T, out, operation, member, prefix string) {
+	t.Helper()
+	assert.Contains(t, out, "dropping foreign index member to prevent CROSSSLOT",
+		"expected warn log emitted by %s; got: %s", operation, out)
+	assert.Contains(t, out, member, "warn log should name the dropped member")
+	assert.Contains(t, out, operation, "warn log should name the operation")
+	assert.Contains(t, out, strings.TrimSuffix(prefix, ":"),
+		"warn log should reference the expected prefix")
+}
+
+// TestForeignMembersFilteredFromIndexOps verifies that the three multi-key
+// SMEMBERS-fed operations (GetAllUpstreamTokens, DeleteUpstreamTokens,
+// GetLatestUpstreamTokensForUser) drop foreign members before MGet/Del so a
+// stray un-prefixed entry in an index set cannot escalate into a CROSSSLOT
+// failure on Redis Cluster, and emit a warn log naming the dropped member.
+//
+// This test uses slog.SetDefault (process-global) and therefore does not run
+// in parallel.
+func TestForeignMembersFilteredFromIndexOps(t *testing.T) { //nolint:paralleltest // captures slog default
+	storage, mr := newTestRedisStorage(t)
+	t.Cleanup(func() {
+		_ = storage.Close()
+		mr.Close()
+	})
+	ctx := context.Background()
+
+	tokens := &UpstreamTokens{
+		ProviderID:   "github",
+		AccessToken:  "real-access",
+		RefreshToken: "real-refresh",
+		UserID:       "user-A",
+		ExpiresAt:    time.Now().Add(time.Hour),
+	}
+	require.NoError(t, storage.StoreUpstreamTokens(ctx, "session-X", "github", tokens))
+
+	sessionIdxKey := redisSetKey(storage.keyPrefix, KeyTypeUpstreamIdx, "session-X")
+	userIdxKey := redisSetKey(storage.keyPrefix, KeyTypeUserUpstream, "user-A")
+	const foreignMember = "other-tenant:auth:{ns:other}:upstream:s:p"
+	mr.SAdd(sessionIdxKey, foreignMember)
+	mr.SAdd(userIdxKey, foreignMember)
+
+	t.Run("GetAllUpstreamTokens", func(t *testing.T) {
+		buf := captureWarnLogs(t)
+		got, err := storage.GetAllUpstreamTokens(ctx, "session-X")
+		require.NoError(t, err)
+		require.Len(t, got, 1)
+		require.Contains(t, got, "github")
+		assert.Equal(t, "real-access", got["github"].AccessToken)
+		assertForeignDropWarn(t, buf.String(), "GetAllUpstreamTokens", foreignMember, storage.keyPrefix)
+	})
+
+	t.Run("GetLatestUpstreamTokensForUser", func(t *testing.T) {
+		buf := captureWarnLogs(t)
+		got, err := storage.GetLatestUpstreamTokensForUser(ctx, "user-A", "github")
+		require.NoError(t, err)
+		require.NotNil(t, got)
+		assert.Equal(t, "real-refresh", got.RefreshToken)
+		assertForeignDropWarn(t, buf.String(), "GetLatestUpstreamTokensForUser", foreignMember, storage.keyPrefix)
+	})
+
+	// DeleteUpstreamTokens runs last because it removes the index set.
+	t.Run("DeleteUpstreamTokens", func(t *testing.T) {
+		buf := captureWarnLogs(t)
+		require.NoError(t, storage.DeleteUpstreamTokens(ctx, "session-X"))
+		assertForeignDropWarn(t, buf.String(), "DeleteUpstreamTokens", foreignMember, storage.keyPrefix)
+	})
+}
+
+func TestFilterIndexMembersByPrefix(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name        string
+		prefix      string
+		members     []string
+		wantKept    []string
+		wantDropped []string
+	}{
+		{
+			name:        "all members share prefix",
+			prefix:      "thv:auth:{ns:name}:",
+			members:     []string{"thv:auth:{ns:name}:upstream:s1:p1", "thv:auth:{ns:name}:upstream:s1:p2"},
+			wantKept:    []string{"thv:auth:{ns:name}:upstream:s1:p1", "thv:auth:{ns:name}:upstream:s1:p2"},
+			wantDropped: nil,
+		},
+		{
+			name:        "stray un-prefixed member is dropped",
+			prefix:      "thv:auth:{ns:name}:",
+			members:     []string{"thv:auth:{ns:name}:upstream:s1:p1", "legacy-key", "thv:auth:{ns:name}:upstream:s1:p2"},
+			wantKept:    []string{"thv:auth:{ns:name}:upstream:s1:p1", "thv:auth:{ns:name}:upstream:s1:p2"},
+			wantDropped: []string{"legacy-key"},
+		},
+		{
+			name:        "different-tenant member is dropped",
+			prefix:      "thv:auth:{ns-a:srv}:",
+			members:     []string{"thv:auth:{ns-a:srv}:upstream:s1:p1", "thv:auth:{ns-b:srv}:upstream:s1:p1"},
+			wantKept:    []string{"thv:auth:{ns-a:srv}:upstream:s1:p1"},
+			wantDropped: []string{"thv:auth:{ns-b:srv}:upstream:s1:p1"},
+		},
+		{
+			name:        "empty input",
+			prefix:      "thv:auth:{ns:name}:",
+			members:     nil,
+			wantKept:    nil,
+			wantDropped: nil,
+		},
+		{
+			name:        "all members dropped",
+			prefix:      "thv:auth:{ns:name}:",
+			members:     []string{"foo", "bar"},
+			wantKept:    nil,
+			wantDropped: []string{"foo", "bar"},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			kept, dropped := filterIndexMembersByPrefix(tc.prefix, tc.members)
+			assert.Equal(t, tc.wantKept, kept, "kept slice mismatch")
+			assert.Equal(t, tc.wantDropped, dropped, "dropped slice mismatch")
+		})
+	}
+}