MUL-7281: support universal Redis clients and cluster mode (#8300)

* feat(redis): support standalone and cluster clients

Co-authored-by: multica-agent <[email protected]>

* fix(redis): use one global Redis configuration

Co-authored-by: multica-agent <[email protected]>

---------

Co-authored-by: Sol-Boy <[email protected]>
Co-authored-by: multica-agent <[email protected]>

.env.example

@@ -399,13 +399,29 @@ CORS_ALLOWED_ORIGINS=
 # (/auth/send-code, /auth/verify-code, /auth/google). Backed by Redis.
 # When REDIS_URL is unset the limiter is a no-op (fail-open) and the
 # backend logs "auth rate limiting disabled: REDIS_URL not configured" at
-# startup. By default the same REDIS_URL is reused by the realtime fan-out
-# hub, the PAT cache, and the daemon-token cache.
+# startup. All Redis-backed features share this one Redis configuration,
+# including realtime fan-out, channel leases, rate limits, and token caches.
 # REDIS_URL=redis://localhost:6379/0
-# Recommended for production: isolate realtime streams from request-path
-# caches and lease state. Requires Redis/Valkey with XTRIM MINID support
-# (Redis 6.2+). When unset, the relay falls back to REDIS_URL.
-# REALTIME_RELAY_REDIS_URL=redis://realtime-redis:6379/0
+# Cluster deployments support only database 0 and sharded realtime relay mode.
+# The server rejects legacy or dual relay mode when this switch is enabled.
+# REDIS_CLUSTER_MODE=false
+
+# Common Redis configurations (choose one):
+#
+# 1. Standalone Redis
+# REDIS_URL=redis://:[email protected]:6379/0
+# REDIS_CLUSTER_MODE=false
+#
+# 2. AWS ElastiCache Serverless (single configuration endpoint + required TLS)
+# The rediss:// scheme is required to enable TLS.
+# REDIS_URL=rediss://:[email protected]:6379/0
+# REDIS_CLUSTER_MODE=true
+#
+# 3. Native multi-node Redis Cluster
+# Separate node addresses between @ and / with commas.
+# REDIS_URL=redis://:[email protected]:6379,10.0.0.2:6379,10.0.0.3:6379/0
+# REDIS_CLUSTER_MODE=true
+# Realtime relay requires Redis/Valkey with XTRIM MINID support (Redis 6.2+).
 # Relay retention defaults: 8 shards, 2000 entries per shard, 5m replay, and
 # exact time trimming after 10m. MAXLEN is the immediate operational pressure
 # valve; size it from observed entry bytes, event rate, and the Redis memory
@@ -433,19 +449,14 @@ CORS_ALLOWED_ORIGINS=
 # REDIS_DISABLE_CLIENT_NAME=true
 
 # ==================== Channel WebSocket leases ====================
-# Multi-replica production should use Redis. The channel supervisor gets a
-# dedicated Redis client/pool and fails closed (does not start connections) if
-# its Redis URL, Ping, or the atomic lease scripts are unavailable at startup.
+# Multi-replica production should use Redis. The channel supervisor gets its
+# own client/pool from REDIS_URL and fails closed (does not start connections)
+# if Redis, Ping, or the atomic lease scripts are unavailable at startup.
 # PostgreSQL remains available for self-hosted single-node use and rollback.
 # Switch every API replica together; there is intentionally no dual-lock mode.
 # CHANNEL_WS_LEASE_BACKEND=redis
-# Strongly recommended in production: point this at a lease-only Redis/Valkey
-# instance. When unset it falls back to REDIS_URL (acceptable for local/small
-# self-hosted deployments, but it shares memory and eviction policy with
-# caches, rate limits, and realtime state).
-# CHANNEL_WS_LEASE_REDIS_URL=redis://lease-redis:6379/0
-# REQUIRED operational policy for the lease instance: `maxmemory-policy
-# noeviction`. Enable HA plus persistence appropriate to the provider, and
+# REQUIRED operational policy for the global Redis configured by REDIS_URL:
+# `maxmemory-policy noeviction`. Enable HA plus persistence appropriate to the provider, and
 # alert on memory/capacity, evictions (must stay zero), availability, and
 # command latency. Silent lease-key eviction can interrupt channel connections.
 # Required in Redis mode. Use a stable deployment identifier so environments

apps/docs/content/docs/environment-variables.ja.mdx

@@ -156,7 +156,8 @@ allowlist の正確な判定順序は[ログインとサインアップ](/auth-s
 
 | 変数 | デフォルト | 説明 |
 |---|---|---|
-| `REDIS_URL` | 空 | 共有レート制限、リアルタイムイベント、token cache に使用。未設定の場合、リアルタイムイベントと招待制限はプロセス内メモリにフォールバックし、認証レート制限は無効になります |
+| `REDIS_URL` | 空 | 共有レート制限、リアルタイムイベント、チャンネル WebSocket リース、token cache など、すべての Redis 機能が使用する唯一の接続 URL |
+| `REDIS_CLUSTER_MODE` | `false` | `true` にすると、`REDIS_URL` が単一の設定エンドポイントでも Redis Cluster クライアントを使用します。クラスターモードではデータベース 0 と `REALTIME_RELAY_MODE=sharded` が必要です |
 | `REDIS_DISABLE_CLIENT_NAME` | `false` | マネージド Redis が `CLIENT SETNAME` を禁止している場合に `true` を設定 |
 | `RATE_LIMIT_AUTH` | `5` | 認証コード送信または Google ログイン開始の、IP あたり毎分の回数 |
 | `RATE_LIMIT_AUTH_VERIFY` | `20` | 認証コード検証の、IP あたり毎分の回数 |

apps/docs/content/docs/environment-variables.ko.mdx

@@ -156,7 +156,8 @@ allowlist의 정확한 판단 순서는 [로그인과 가입](/auth-setup#가입
 
 | 변수 | 기본값 | 설명 |
 |---|---|---|
-| `REDIS_URL` | 비어 있음 | 공유 rate limit, 실시간 이벤트, token cache에 사용. 설정하지 않으면 실시간 이벤트와 초대 제한은 프로세스 메모리로 fallback하고 인증 rate limit은 비활성화 |
+| `REDIS_URL` | 비어 있음 | 공유 rate limit, 실시간 이벤트, 채널 WebSocket lease, token cache 등 모든 Redis 기능이 사용하는 유일한 연결 URL |
+| `REDIS_CLUSTER_MODE` | `false` | `true`로 설정하면 `REDIS_URL`이 단일 구성 엔드포인트여도 Redis Cluster 클라이언트를 사용합니다. 클러스터 모드에서는 데이터베이스 0과 `REALTIME_RELAY_MODE=sharded`가 필요합니다 |
 | `REDIS_DISABLE_CLIENT_NAME` | `false` | 관리형 Redis가 `CLIENT SETNAME`을 금지하면 `true`로 설정 |
 | `RATE_LIMIT_AUTH` | `5` | IP당 1분에 인증 코드 전송 또는 Google 로그인 시작 허용 횟수 |
 | `RATE_LIMIT_AUTH_VERIFY` | `20` | IP당 1분에 인증 코드 검증 허용 횟수 |

apps/docs/content/docs/environment-variables.mdx

@@ -156,7 +156,8 @@ Use `ATTACHMENT_DOWNLOAD_MODE=proxy` when the endpoint — an internal MinIO, fo
 
 | Variable | Default | Description |
 |---|---|---|
-| `REDIS_URL` | empty | Used for shared rate limiting, realtime events, and the token cache; when unset, realtime events and invitation limits fall back to in-process memory, while auth rate limiting is off |
+| `REDIS_URL` | empty | The one connection URL used by all Redis-backed features, including shared rate limiting, realtime events, channel WebSocket leases, and token caches |
+| `REDIS_CLUSTER_MODE` | `false` | Forces the Redis Cluster client, including for a single configuration endpoint; cluster mode requires database 0 and `REALTIME_RELAY_MODE=sharded` |
 | `REDIS_DISABLE_CLIENT_NAME` | `false` | Set `true` when a managed Redis blocks `CLIENT SETNAME` |
 | `RATE_LIMIT_AUTH` | `5` | Per-IP requests per minute to send a verification code or start Google sign-in |
 | `RATE_LIMIT_AUTH_VERIFY` | `20` | Per-IP code verifications per minute |

apps/docs/content/docs/environment-variables.zh.mdx

@@ -156,7 +156,8 @@ MULTICA_PUBLIC_URL=https://api.multica.example.com
 
 | 变量 | 默认值 | 说明 |
 |---|---|---|
-| `REDIS_URL` | 空 | 用于共享限流、实时事件和 token cache；未设置时实时事件和邀请限流回退到进程内存，认证限流不生效 |
+| `REDIS_URL` | 空 | 所有 Redis 功能共用的唯一连接地址，包括共享限流、实时事件、频道 WebSocket 租约和 token cache |
+| `REDIS_CLUSTER_MODE` | `false` | 设为 `true` 后强制使用 Redis Cluster 客户端，单配置地址也适用；集群模式只能使用数据库 0，且要求 `REALTIME_RELAY_MODE=sharded` |
 | `REDIS_DISABLE_CLIENT_NAME` | `false` | 托管 Redis 禁止 `CLIENT SETNAME` 时设为 `true` |
 | `RATE_LIMIT_AUTH` | `5` | 单 IP 每分钟发送验证码或发起 Google 登录的次数 |
 | `RATE_LIMIT_AUTH_VERIFY` | `20` | 单 IP 每分钟验证验证码的次数 |

docker-compose.selfhost.yml

@@ -67,6 +67,8 @@ services:
       DATABASE_REPLICA_URL: ${DATABASE_REPLICA_URL:-}
       DATABASE_REPLICA_MAX_CONNS: ${DATABASE_REPLICA_MAX_CONNS:-}
       DATABASE_REPLICA_MIN_CONNS: ${DATABASE_REPLICA_MIN_CONNS:-}
+      REDIS_URL: ${REDIS_URL:-}
+      REDIS_CLUSTER_MODE: ${REDIS_CLUSTER_MODE:-false}
       PORT: "8080"
       METRICS_ADDR: ${METRICS_ADDR:-}
       JWT_SECRET: ${JWT_SECRET:?JWT_SECRET must be set to a strong random value — generate one with 'openssl rand -hex 32'}

server/cmd/server/main.go

@@ -17,6 +17,7 @@ import (
 	"github.com/multica-ai/multica/server/internal/analytics"
 	"github.com/multica-ai/multica/server/internal/auth"
 	"github.com/multica-ai/multica/server/internal/daemonws"
+	"github.com/multica-ai/multica/server/internal/database"
 	"github.com/multica-ai/multica/server/internal/dbreader"
 	"github.com/multica-ai/multica/server/internal/dbstartup"
 	"github.com/multica-ai/multica/server/internal/events"
@@ -39,14 +40,14 @@ var (
 	commit  = "unknown"
 )
 
-func newNamedRedisClient(base *redis.Options, suffix string) *redis.Client {
+func newNamedRedisClient(base *redis.UniversalOptions, suffix string) redis.UniversalClient {
 	opts := *base
 	if envBool("REDIS_DISABLE_CLIENT_NAME", false) {
 		opts.ClientName = ""
 	} else {
 		opts.ClientName = redisClientName(opts.ClientName, suffix)
 	}
-	return redis.NewClient(&opts)
+	return redis.NewUniversalClient(&opts)
 }
 
 // newClaimRedisClient is a named client that honours its callers' context
@@ -63,7 +64,7 @@ func newNamedRedisClient(base *redis.Options, suffix string) *redis.Client {
 // Hence a dedicated client rather than the flag on the shared relay client:
 // setting it there would change the timeout behaviour of every publish that
 // runs through it, which is a far wider blast radius than this store needs.
-func newClaimRedisClient(base *redis.Options, suffix string) *redis.Client {
+func newClaimRedisClient(base *redis.UniversalOptions, suffix string) redis.UniversalClient {
 	opts := *base
 	opts.ContextTimeoutEnabled = true
 	return newNamedRedisClient(&opts, suffix)
@@ -79,21 +80,7 @@ func redisClientName(existing, suffix string) string {
 	return "multica-api:" + suffix
 }
 
-func channelLeaseRedisURLFromEnv() string {
-	if dedicated := strings.TrimSpace(os.Getenv("CHANNEL_WS_LEASE_REDIS_URL")); dedicated != "" {
-		return dedicated
-	}
-	return strings.TrimSpace(os.Getenv("REDIS_URL"))
-}
-
-func realtimeRelayRedisURLFromEnv() string {
-	if dedicated := strings.TrimSpace(os.Getenv("REALTIME_RELAY_REDIS_URL")); dedicated != "" {
-		return dedicated
-	}
-	return strings.TrimSpace(os.Getenv("REDIS_URL"))
-}
-
-func closeRedisClient(label string, client *redis.Client) {
+func closeRedisClient(label string, client redis.UniversalClient) {
 	if client == nil {
 		return
 	}
@@ -135,6 +122,13 @@ func realtimeRelayModeFromEnv() string {
 	}
 }
 
+func validateRealtimeRelayMode(mode string, clusterMode bool) error {
+	if clusterMode && mode != "sharded" {
+		return fmt.Errorf("REALTIME_RELAY_MODE=%s is incompatible with REDIS_CLUSTER_MODE=true; use sharded mode", mode)
+	}
+	return nil
+}
+
 func envPositiveInt(name string, def int) int {
 	raw := os.Getenv(name)
 	if raw == "" {
@@ -430,17 +424,18 @@ func main() {
 	// is the sole broadcaster and the server stays single-node (legacy).
 	// Runtime local-skill stores and realtime relay traffic use separate Redis
 	// clients so blocking stream consumers cannot starve request-path Redis
-	// operations. Channel leases are initialized separately below so production
-	// can point them at a dedicated no-eviction Redis instance.
+	// operations. Channel leases are initialized separately below from the same
+	// global Redis configuration.
 	relayCtx, relayCancel := context.WithCancel(context.Background())
 	var broadcaster realtime.Broadcaster = hub
-	var storeRedis *redis.Client
-	var channelLeaseRedis *redis.Client
-	var relayWriteRedis *redis.Client
-	var wecomClaimRedis *redis.Client
-	var relayReadRedis *redis.Client
-	var shardedReadRedis *redis.Client
-	var legacyReadRedis *redis.Client
+	var storeRedis redis.UniversalClient
+	var storeRedisPoolSize int
+	var channelLeaseRedis redis.UniversalClient
+	var relayWriteRedis redis.UniversalClient
+	var wecomClaimRedis redis.UniversalClient
+	var relayReadRedis redis.UniversalClient
+	var shardedReadRedis redis.UniversalClient
+	var legacyReadRedis redis.UniversalClient
 	var relay realtime.ManagedRelay
 	// stopRelay halts the relay readers and drains the WeCom dispatcher. It is
 	// called from the shutdown BODY, before the channel supervisor is torn
@@ -476,25 +471,34 @@ func main() {
 		closeRedisClient("store", storeRedis)
 	}()
 	sharedRedisURL := strings.TrimSpace(os.Getenv("REDIS_URL"))
-	relayRedisURL := realtimeRelayRedisURLFromEnv()
-	if (sharedRedisURL != "" || relayRedisURL != "") && envBool("REDIS_DISABLE_CLIENT_NAME", false) {
+	redisClusterMode := envBool("REDIS_CLUSTER_MODE", false)
+	// Main parses shared options instead of using database.NewRedisClient so it
+	// can create separate role-specific pools. Request-path clients must also
+	// survive a transient startup outage; relay and lease components own their
+	// existing bounded readiness probes and failure policies.
+	if sharedRedisURL != "" && envBool("REDIS_DISABLE_CLIENT_NAME", false) {
 		slog.Info("redis: CLIENT SETNAME disabled (REDIS_DISABLE_CLIENT_NAME=true) for managed Redis compatibility")
 	}
 	if sharedRedisURL != "" {
-		if opts, err := redis.ParseURL(sharedRedisURL); err != nil {
+		if opts, err := database.NewRedisOptions(database.RedisConfig{URL: sharedRedisURL, ClusterMode: redisClusterMode}); err != nil {
 			slog.Error("invalid REDIS_URL — request-path Redis features disabled", "error", err)
 		} else {
 			storeRedis = newNamedRedisClient(opts, "store")
+			storeRedisPoolSize = opts.PoolSize
 		}
 	}
-	if relayRedisURL != "" {
-		opts, err := redis.ParseURL(relayRedisURL)
+	if sharedRedisURL != "" {
+		opts, err := database.NewRedisOptions(database.RedisConfig{URL: sharedRedisURL, ClusterMode: redisClusterMode})
 		if err != nil {
-			slog.Error("invalid realtime relay Redis URL — falling back to in-memory hub", "error", err)
+			slog.Error("invalid REDIS_URL — realtime relay falling back to in-memory hub", "error", err)
 		} else {
+			relayMode := realtimeRelayModeFromEnv()
+			if err := validateRealtimeRelayMode(relayMode, redisClusterMode); err != nil {
+				slog.Error("invalid realtime relay configuration", "error", err)
+				os.Exit(1)
+			}
 			relayWriteRedis = newNamedRedisClient(opts, "realtime-write")
 
-			relayMode := realtimeRelayModeFromEnv()
 			relayConfig := shardedRelayConfigFromEnv()
 			switch relayMode {
 			case "legacy":
@@ -544,15 +548,10 @@ func main() {
 			// silently misses it.
 			relay.Start(relayCtx)
 			broadcaster = realtime.NewDualWriteBroadcaster(hub, relay)
-			storePoolSize := 0
-			if storeRedis != nil {
-				storePoolSize = storeRedis.Options().PoolSize
-			}
 			slog.Info(
 				"realtime: Redis relay enabled",
 				"node_id", relay.NodeID(),
 				"mode", relayMode,
-				"dedicated_instance", strings.TrimSpace(os.Getenv("REALTIME_RELAY_REDIS_URL")) != "",
 				"shards", relayConfig.Shards,
 				"stream_max_len", relayConfig.StreamMaxLen,
 				"replay_grace", relayConfig.ReplayGrace.String(),
@@ -561,19 +560,18 @@ func main() {
 				"stream_ttl_enabled", relayConfig.StreamTTLEnabled,
 				"xread_count", relayConfig.ReadCount,
 				"xread_block", relayConfig.ReadBlock.String(),
-				"store_pool_size", storePoolSize,
+				"store_pool_size", storeRedisPoolSize,
 				"realtime_write_pool_size", opts.PoolSize,
 				"realtime_read_pool_size", opts.PoolSize,
 			)
 		}
 	} else {
-		slog.Info("realtime: REDIS_URL and REALTIME_RELAY_REDIS_URL are unset — using in-memory hub (single-node mode)")
+		slog.Info("realtime: REDIS_URL is unset — using in-memory hub (single-node mode)")
 	}
 	if strings.EqualFold(strings.TrimSpace(os.Getenv("CHANNEL_WS_LEASE_BACKEND")), "redis") {
-		leaseRedisURL := channelLeaseRedisURLFromEnv()
-		if leaseRedisURL == "" {
-			slog.Error("channel leases: CHANNEL_WS_LEASE_REDIS_URL and REDIS_URL are unset")
-		} else if opts, err := redis.ParseURL(leaseRedisURL); err != nil {
+		if sharedRedisURL == "" {
+			slog.Error("channel leases: REDIS_URL is unset")
+		} else if opts, err := database.NewRedisOptions(database.RedisConfig{URL: sharedRedisURL, ClusterMode: redisClusterMode}); err != nil {
 			slog.Error("channel leases: invalid Redis URL; supervisor will fail closed", "error", err)
 		} else {
 			channelLeaseRedis = newNamedRedisClient(opts, "channel-lease")

server/cmd/server/main_test.go

@@ -32,35 +32,26 @@ func TestRedisClientName(t *testing.T) {
 	}
 }
 
-func TestChannelLeaseRedisURLFromEnvPrefersDedicatedInstance(t *testing.T) {
-	t.Setenv("REDIS_URL", "redis://shared:6379/0")
-	t.Setenv("CHANNEL_WS_LEASE_REDIS_URL", "redis://leases:6379/0")
-	if got := channelLeaseRedisURLFromEnv(); got != "redis://leases:6379/0" {
-		t.Fatalf("channel lease Redis URL = %q", got)
+func TestValidateRealtimeRelayMode(t *testing.T) {
+	tests := []struct {
+		name        string
+		mode        string
+		clusterMode bool
+		wantErr     bool
+	}{
+		{name: "standalone legacy", mode: "legacy"},
+		{name: "standalone dual", mode: "dual"},
+		{name: "cluster sharded", mode: "sharded", clusterMode: true},
+		{name: "cluster legacy", mode: "legacy", clusterMode: true, wantErr: true},
+		{name: "cluster dual", mode: "dual", clusterMode: true, wantErr: true},
 	}
-}
-
-func TestChannelLeaseRedisURLFromEnvFallsBackToSharedRedis(t *testing.T) {
-	t.Setenv("REDIS_URL", "redis://shared:6379/0")
-	t.Setenv("CHANNEL_WS_LEASE_REDIS_URL", "")
-	if got := channelLeaseRedisURLFromEnv(); got != "redis://shared:6379/0" {
-		t.Fatalf("channel lease Redis URL = %q", got)
-	}
-}
-
-func TestRealtimeRelayRedisURLFromEnvPrefersDedicatedInstance(t *testing.T) {
-	t.Setenv("REDIS_URL", "redis://shared:6379/0")
-	t.Setenv("REALTIME_RELAY_REDIS_URL", " redis://relay:6379/0 ")
-	if got := realtimeRelayRedisURLFromEnv(); got != "redis://relay:6379/0" {
-		t.Fatalf("realtime relay Redis URL = %q", got)
-	}
-}
-
-func TestRealtimeRelayRedisURLFromEnvFallsBackToSharedRedis(t *testing.T) {
-	t.Setenv("REDIS_URL", " redis://shared:6379/0 ")
-	t.Setenv("REALTIME_RELAY_REDIS_URL", "")
-	if got := realtimeRelayRedisURLFromEnv(); got != "redis://shared:6379/0" {
-		t.Fatalf("realtime relay Redis URL = %q", got)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := validateRealtimeRelayMode(tt.mode, tt.clusterMode)
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("validateRealtimeRelayMode(%q, %t) error = %v, wantErr %t", tt.mode, tt.clusterMode, err, tt.wantErr)
+			}
+		})
 	}
 }
 
@@ -120,11 +111,11 @@ func TestShardedRelayConfigFromEnvNormalizesUnsafeOverrides(t *testing.T) {
 
 func TestNewNamedRedisClient_SetsClientName(t *testing.T) {
 	t.Setenv("REDIS_DISABLE_CLIENT_NAME", "")
-	base := &redis.Options{Addr: "localhost:6379"}
+	base := &redis.UniversalOptions{Addrs: []string{"localhost:6379"}}
 	client := newNamedRedisClient(base, "store")
 	defer client.Close()
 
-	opts := client.Options()
+	opts := client.(*redis.Client).Options()
 	if opts.ClientName != "multica-api:store" {
 		t.Errorf("ClientName = %q, want %q", opts.ClientName, "multica-api:store")
 	}
@@ -132,11 +123,11 @@ func TestNewNamedRedisClient_SetsClientName(t *testing.T) {
 
 func TestNewNamedRedisClient_DisableClientName(t *testing.T) {
 	t.Setenv("REDIS_DISABLE_CLIENT_NAME", "true")
-	base := &redis.Options{Addr: "localhost:6379"}
+	base := &redis.UniversalOptions{Addrs: []string{"localhost:6379"}}
 	client := newNamedRedisClient(base, "store")
 	defer client.Close()
 
-	opts := client.Options()
+	opts := client.(*redis.Client).Options()
 	if opts.ClientName != "" {
 		t.Errorf("ClientName = %q, want empty when REDIS_DISABLE_CLIENT_NAME=true", opts.ClientName)
 	}
@@ -145,11 +136,11 @@ func TestNewNamedRedisClient_DisableClientName(t *testing.T) {
 func TestNewNamedRedisClient_DisableClientName_ClearsPreExistingName(t *testing.T) {
 	t.Setenv("REDIS_DISABLE_CLIENT_NAME", "true")
 	// Simulate REDIS_URL with ?client_name=foo — ParseURL sets ClientName.
-	base := &redis.Options{Addr: "localhost:6379", ClientName: "foo"}
+	base := &redis.UniversalOptions{Addrs: []string{"localhost:6379"}, ClientName: "foo"}
 	client := newNamedRedisClient(base, "store")
 	defer client.Close()
 
-	opts := client.Options()
+	opts := client.(*redis.Client).Options()
 	if opts.ClientName != "" {
 		t.Errorf("ClientName = %q, want empty: REDIS_DISABLE_CLIENT_NAME must clear pre-existing name from URL", opts.ClientName)
 	}
@@ -157,11 +148,11 @@ func TestNewNamedRedisClient_DisableClientName_ClearsPreExistingName(t *testing.
 
 func TestNewNamedRedisClient_DisableClientName_InvalidValue(t *testing.T) {
 	t.Setenv("REDIS_DISABLE_CLIENT_NAME", "not-a-bool")
-	base := &redis.Options{Addr: "localhost:6379"}
+	base := &redis.UniversalOptions{Addrs: []string{"localhost:6379"}}
 	client := newNamedRedisClient(base, "store")
 	defer client.Close()
 
-	opts := client.Options()
+	opts := client.(*redis.Client).Options()
 	// Invalid value falls back to default (false), so ClientName IS set
 	if opts.ClientName != "multica-api:store" {
 		t.Errorf("ClientName = %q, want %q (invalid env should fall back to naming enabled)", opts.ClientName, "multica-api:store")

server/cmd/server/router.go

@@ -208,7 +208,7 @@ func normalizeServerVersion(v string) string {
 // path Redis client, not the realtime relay's blocking read client. A nil rdb
 // keeps the default in-memory stores which are fine for single-node dev and
 // tests.
-func NewRouter(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus, analyticsClient analytics.Client, rdb *redis.Client) chi.Router {
+func NewRouter(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus, analyticsClient analytics.Client, rdb redis.UniversalClient) chi.Router {
 	r, _ := NewRouterWithOptions(pool, hub, bus, analyticsClient, rdb, RouterOptions{})
 	return r
 }
@@ -219,7 +219,7 @@ type RouterOptions struct {
 	ChannelLeaseMetrics *obsmetrics.ChannelLeaseMetrics
 	// ChannelLeaseRedis is a dedicated non-blocking Redis client/pool. It is
 	// required only when CHANNEL_WS_LEASE_BACKEND=redis.
-	ChannelLeaseRedis *redis.Client
+	ChannelLeaseRedis redis.UniversalClient
 	// WecomRelay is the realtime relay, seen through the two halves the WeCom
 	// adapter needs: publish a reply to the other replicas, and register as
 	// the consumer that delivers the ones they publish. Nil on a deployment
@@ -279,7 +279,7 @@ func buildChannelSupervisor(
 		leases = postgresLeases
 	case "redis":
 		if opts.ChannelLeaseRedis == nil {
-			slog.Error("channel engine: Redis lease backend selected but CHANNEL_WS_LEASE_REDIS_URL/REDIS_URL is missing or invalid; supervisor disabled")
+			slog.Error("channel engine: Redis lease backend selected but REDIS_URL is missing or invalid; supervisor disabled")
 			return nil
 		}
 		namespace := strings.TrimSpace(os.Getenv("CHANNEL_WS_LEASE_NAMESPACE"))
@@ -394,7 +394,7 @@ func seatCapacityExecutor(cloudURL string) seatcapacity.Executor {
 // context, calling Wait on shutdown) use the returned handler;
 // callers that only need the HTTP handler (tests, the simple
 // NewRouter shim) discard the second value.
-func NewRouterWithOptions(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus, analyticsClient analytics.Client, rdb *redis.Client, opts RouterOptions) (chi.Router, *handler.Handler) {
+func NewRouterWithOptions(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus, analyticsClient analytics.Client, rdb redis.UniversalClient, opts RouterOptions) (chi.Router, *handler.Handler) {
 	queries := db.New(pool)
 	emailSvc := service.NewEmailService()
 	daemonHub := opts.DaemonHub

server/cmd/server/wecom_claim_redis_wiring_test.go

@@ -22,11 +22,12 @@ import (
 // dedupe store is handed the right one is guarded at runtime instead, by the
 // warning NewRedisDedupe logs for a client that ignores deadlines.
 func TestClaimRedisClientHonoursContextDeadlines(t *testing.T) {
-	base := &redis.Options{Addr: "localhost:6379", ClientName: "multica", ReadTimeout: 3 * time.Second}
+	base := &redis.UniversalOptions{Addrs: []string{"localhost:6379"}, ClientName: "multica", ReadTimeout: 3 * time.Second}
 
 	claim := newClaimRedisClient(base, "wecom-claim")
 	defer claim.Close()
-	if !claim.Options().ContextTimeoutEnabled {
+	claimOptions := claim.(*redis.Client).Options()
+	if !claimOptions.ContextTimeoutEnabled {
 		t.Fatal("the claim client ignores context deadlines: ClaimBudget and the drain budget " +
 			"would be numbers nothing enforces")
 	}
@@ -35,16 +36,17 @@ func TestClaimRedisClientHonoursContextDeadlines(t *testing.T) {
 	// scoped change rather than a fleet-wide one.
 	shared := newNamedRedisClient(base, "realtime-write")
 	defer shared.Close()
-	if shared.Options().ContextTimeoutEnabled {
+	sharedOptions := shared.(*redis.Client).Options()
+	if sharedOptions.ContextTimeoutEnabled {
 		t.Fatal("the shared relay client now enforces context deadlines too: that is a wider " +
 			"blast radius than the claim store asked for")
 	}
 
 	// Everything else about the client is still the deployment's.
-	if got := claim.Options().ReadTimeout; got != base.ReadTimeout {
+	if got := claimOptions.ReadTimeout; got != base.ReadTimeout {
 		t.Fatalf("ReadTimeout = %v, want the configured %v", got, base.ReadTimeout)
 	}
-	if got := claim.Options().ClientName; got != "multica:wecom-claim" {
+	if got := claimOptions.ClientName; got != "multica:wecom-claim" {
 		t.Fatalf("ClientName = %q, want the suffixed name", got)
 	}
 	if base.ContextTimeoutEnabled {

server/internal/auth/cloud_pat.go

@@ -168,7 +168,7 @@ type OwnerLookupFunc func(ctx context.Context, ownerID string) (bool, error)
 type CloudPATVerifier struct {
 	baseURL string
 	http    *http.Client
-	rdb     *redis.Client // may be nil — disables caching
+	rdb     redis.UniversalClient // may be nil — disables caching
 }
 
 // CloudPATVerifierConfig assembles the dependencies for
@@ -190,7 +190,7 @@ type CloudPATVerifierConfig struct {
 	// Redis backs the positive-result cache. Nil disables caching —
 	// every Verify call hits Fleet. Same nil-safe contract as
 	// PATCache / DaemonTokenCache.
-	Redis *redis.Client
+	Redis redis.UniversalClient
 }
 
 // NewCloudPATVerifier returns a verifier for cfg.FleetBaseURL. If the

server/internal/auth/daemon_token_cache.go

@@ -29,13 +29,13 @@ type DaemonTokenIdentity struct {
 // or reports a cache miss, so single-node dev / tests with no REDIS_URL
 // degrade cleanly to direct DB lookups.
 type DaemonTokenCache struct {
-	rdb *redis.Client
+	rdb redis.UniversalClient
 }
 
 // NewDaemonTokenCache returns a cache backed by rdb. Pass nil to disable
 // caching; the returned *DaemonTokenCache is safe to call but never hits
 // Redis.
-func NewDaemonTokenCache(rdb *redis.Client) *DaemonTokenCache {
+func NewDaemonTokenCache(rdb redis.UniversalClient) *DaemonTokenCache {
 	if rdb == nil {
 		return nil
 	}

server/internal/auth/membership_cache.go

@@ -32,13 +32,13 @@ const MembershipCacheTTL = 5 * time.Minute
 // A nil *MembershipCache is safe to use — every method becomes a no-op or
 // reports a cache miss, and the caller degrades to direct DB lookups.
 type MembershipCache struct {
-	rdb *redis.Client
+	rdb redis.UniversalClient
 }
 
 // NewMembershipCache returns a cache backed by rdb. Pass nil to disable
 // caching; the returned *MembershipCache is safe to call but never hits
 // Redis.
-func NewMembershipCache(rdb *redis.Client) *MembershipCache {
+func NewMembershipCache(rdb redis.UniversalClient) *MembershipCache {
 	if rdb == nil {
 		return nil
 	}

server/internal/auth/pat_cache.go

@@ -26,12 +26,12 @@ const patCachePrefix = "mul:auth:pat:"
 // to use — every method becomes a no-op or reports a cache miss, and the
 // auth middleware degrades to direct DB lookups.
 type PATCache struct {
-	rdb *redis.Client
+	rdb redis.UniversalClient
 }
 
 // NewPATCache returns a cache backed by rdb. Pass nil to disable caching;
 // the returned *PATCache is safe to call but never hits Redis.
-func NewPATCache(rdb *redis.Client) *PATCache {
+func NewPATCache(rdb redis.UniversalClient) *PATCache {
 	if rdb == nil {
 		return nil
 	}

server/internal/database/redis.go

@@ -0,0 +1,167 @@
+package database
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const redisPingTimeout = 5 * time.Second
+
+// RedisConfig maps the shared Redis connection environment variables.
+type RedisConfig struct {
+	URL         string `env:"REDIS_URL"`
+	ClusterMode bool   `env:"REDIS_CLUSTER_MODE"`
+}
+
+// NewRedisClient builds a standalone or cluster client and verifies that it is
+// reachable before returning it. rediss:// URLs enable TLS through ParseURL.
+func NewRedisClient(cfg RedisConfig) (redis.UniversalClient, error) {
+	opts, err := NewRedisOptions(cfg)
+	if err != nil {
+		return nil, err
+	}
+	client := redis.NewUniversalClient(opts)
+	ctx, cancel := context.WithTimeout(context.Background(), redisPingTimeout)
+	defer cancel()
+	if err := client.Ping(ctx).Err(); err != nil {
+		return nil, errors.Join(fmt.Errorf("ping redis: %w", err), client.Close())
+	}
+	return client, nil
+}
+
+// RedisContextTimeoutEnabled reports the context-deadline setting for clients
+// created by NewUniversalClient. It keeps concrete client inspection inside
+// the initialization package while consumers depend only on UniversalClient.
+func RedisContextTimeoutEnabled(client redis.UniversalClient) (enabled, known bool) {
+	switch client := client.(type) {
+	case *redis.Client:
+		return client.Options().ContextTimeoutEnabled, true
+	case *redis.ClusterClient:
+		return client.Options().ContextTimeoutEnabled, true
+	default:
+		return false, false
+	}
+}
+
+// NewRedisOptions parses the URL once per seed while preserving the connection,
+// pool, authentication, and TLS options understood by go-redis. A comma-separated
+// authority is normalized before ParseURL because net/url rejects multiple ports.
+func NewRedisOptions(cfg RedisConfig) (*redis.UniversalOptions, error) {
+	rawURL := strings.TrimSpace(cfg.URL)
+	if strings.HasPrefix(rawURL, "unix://") {
+		if cfg.ClusterMode {
+			return nil, errors.New("redis cluster mode does not support unix socket URLs")
+		}
+		opts, err := redis.ParseURL(rawURL)
+		if err != nil {
+			return nil, fmt.Errorf("parse redis URL: %w", err)
+		}
+		return universalOptions(opts, []string{opts.Addr}, false), nil
+	}
+
+	seedURLs, err := redisSeedURLs(rawURL)
+	if err != nil {
+		return nil, err
+	}
+	if len(seedURLs) > 1 && !cfg.ClusterMode {
+		return nil, errors.New("REDIS_CLUSTER_MODE must be true when REDIS_URL contains multiple addresses")
+	}
+
+	var base *redis.Options
+	addrs := make([]string, 0, len(seedURLs))
+	for _, seedURL := range seedURLs {
+		opts, err := redis.ParseURL(seedURL)
+		if err != nil {
+			return nil, fmt.Errorf("parse redis URL: %w", err)
+		}
+		if base == nil {
+			base = opts
+		}
+		addrs = append(addrs, opts.Addr)
+	}
+	if base == nil {
+		return nil, errors.New("REDIS_URL does not contain a Redis address")
+	}
+	if cfg.ClusterMode && base.DB != 0 {
+		return nil, errors.New("REDIS_URL database must be 0 in cluster mode")
+	}
+	return universalOptions(base, addrs, cfg.ClusterMode), nil
+}
+
+func redisSeedURLs(rawURL string) ([]string, error) {
+	schemeEnd := strings.Index(rawURL, "://")
+	if schemeEnd < 0 {
+		return []string{rawURL}, nil
+	}
+	authorityStart := schemeEnd + len("://")
+	authorityEnd := len(rawURL)
+	if end := strings.IndexAny(rawURL[authorityStart:], "/?#"); end >= 0 {
+		authorityEnd = authorityStart + end
+	}
+	hostStart := authorityStart
+	if userinfoEnd := strings.LastIndex(rawURL[authorityStart:authorityEnd], "@"); userinfoEnd >= 0 {
+		hostStart += userinfoEnd + 1
+	}
+
+	seeds := strings.Split(rawURL[hostStart:authorityEnd], ",")
+	seedURLs := make([]string, 0, len(seeds))
+	for _, seed := range seeds {
+		seed = strings.TrimSpace(seed)
+		if seed == "" {
+			return nil, errors.New("REDIS_URL contains an empty Redis address")
+		}
+		seedURLs = append(seedURLs, rawURL[:hostStart]+seed+rawURL[authorityEnd:])
+	}
+	return seedURLs, nil
+}
+
+func universalOptions(opts *redis.Options, addrs []string, clusterMode bool) *redis.UniversalOptions {
+	return &redis.UniversalOptions{
+		Addrs:                        addrs,
+		ClientName:                   opts.ClientName,
+		Dialer:                       opts.Dialer,
+		OnConnect:                    opts.OnConnect,
+		Protocol:                     opts.Protocol,
+		Username:                     opts.Username,
+		Password:                     opts.Password,
+		CredentialsProvider:          opts.CredentialsProvider,
+		CredentialsProviderContext:   opts.CredentialsProviderContext,
+		StreamingCredentialsProvider: opts.StreamingCredentialsProvider,
+		DB:                           opts.DB,
+		MaxRetries:                   opts.MaxRetries,
+		MinRetryBackoff:              opts.MinRetryBackoff,
+		MaxRetryBackoff:              opts.MaxRetryBackoff,
+		DialTimeout:                  opts.DialTimeout,
+		DialerRetries:                opts.DialerRetries,
+		DialerRetryTimeout:           opts.DialerRetryTimeout,
+		ReadTimeout:                  opts.ReadTimeout,
+		WriteTimeout:                 opts.WriteTimeout,
+		ContextTimeoutEnabled:        opts.ContextTimeoutEnabled,
+		ReadBufferSize:               opts.ReadBufferSize,
+		WriteBufferSize:              opts.WriteBufferSize,
+		PoolFIFO:                     opts.PoolFIFO,
+		PoolSize:                     opts.PoolSize,
+		MaxConcurrentDials:           opts.MaxConcurrentDials,
+		PoolTimeout:                  opts.PoolTimeout,
+		MinIdleConns:                 opts.MinIdleConns,
+		MaxIdleConns:                 opts.MaxIdleConns,
+		MaxActiveConns:               opts.MaxActiveConns,
+		ConnMaxIdleTime:              opts.ConnMaxIdleTime,
+		ConnMaxLifetime:              opts.ConnMaxLifetime,
+		ConnMaxLifetimeJitter:        opts.ConnMaxLifetimeJitter,
+		TLSConfig:                    opts.TLSConfig,
+		DisableIndentity:             opts.DisableIndentity,
+		DisableIdentity:              opts.DisableIdentity,
+		IdentitySuffix:               opts.IdentitySuffix,
+		FailingTimeoutSeconds:        opts.FailingTimeoutSeconds,
+		UnstableResp3:                opts.UnstableResp3,
+		PushNotificationProcessor:    opts.PushNotificationProcessor,
+		IsClusterMode:                clusterMode,
+		MaintNotificationsConfig:     opts.MaintNotificationsConfig,
+	}
+}

server/internal/database/redis_test.go

@@ -0,0 +1,133 @@
+package database
+
+import (
+	"os"
+	"testing"
+
+	"github.com/redis/go-redis/v9"
+)
+
+func TestNewRedisOptions(t *testing.T) {
+	tests := []struct {
+		name        string
+		cfg         RedisConfig
+		wantAddrs   []string
+		wantUser    string
+		wantPass    string
+		wantDB      int
+		wantTLS     bool
+		wantSNI     string
+		wantCluster bool
+		wantPool    int
+	}{
+		{
+			name:      "standalone",
+			cfg:       RedisConfig{URL: "redis://app:[email protected]:6380/2?pool_size=17"},
+			wantAddrs: []string{"127.0.0.1:6380"},
+			wantUser:  "app",
+			wantPass:  "secret",
+			wantDB:    2,
+			wantPool:  17,
+		},
+		{
+			name:        "elasticache serverless cluster with TLS",
+			cfg:         RedisConfig{URL: "rediss://:[email protected]:6379/0", ClusterMode: true},
+			wantAddrs:   []string{"cache.example.com:6379"},
+			wantPass:    "secret",
+			wantTLS:     true,
+			wantSNI:     "cache.example.com",
+			wantCluster: true,
+		},
+		{
+			name:        "native multi-node cluster",
+			cfg:         RedisConfig{URL: "redis://:[email protected]:6379,10.0.0.2:6380,cache.internal/0", ClusterMode: true},
+			wantAddrs:   []string{"10.0.0.1:6379", "10.0.0.2:6380", "cache.internal:6379"},
+			wantPass:    "secret",
+			wantCluster: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			opts, err := NewRedisOptions(tt.cfg)
+			if err != nil {
+				t.Fatalf("NewRedisOptions() error = %v", err)
+			}
+			if len(opts.Addrs) != len(tt.wantAddrs) {
+				t.Fatalf("Addrs = %v, want %v", opts.Addrs, tt.wantAddrs)
+			}
+			for i := range tt.wantAddrs {
+				if opts.Addrs[i] != tt.wantAddrs[i] {
+					t.Fatalf("Addrs = %v, want %v", opts.Addrs, tt.wantAddrs)
+				}
+			}
+			if opts.Username != tt.wantUser || opts.Password != tt.wantPass {
+				t.Fatalf("credentials = (%q, %q), want (%q, %q)", opts.Username, opts.Password, tt.wantUser, tt.wantPass)
+			}
+			if opts.DB != tt.wantDB {
+				t.Fatalf("DB = %d, want %d", opts.DB, tt.wantDB)
+			}
+			if (opts.TLSConfig != nil) != tt.wantTLS {
+				t.Fatalf("TLS configured = %v, want %v", opts.TLSConfig != nil, tt.wantTLS)
+			}
+			if tt.wantTLS && opts.TLSConfig.ServerName != tt.wantSNI {
+				t.Fatalf("TLS ServerName = %q, want %q", opts.TLSConfig.ServerName, tt.wantSNI)
+			}
+			if opts.IsClusterMode != tt.wantCluster {
+				t.Fatalf("IsClusterMode = %v, want %v", opts.IsClusterMode, tt.wantCluster)
+			}
+			if opts.PoolSize != tt.wantPool {
+				t.Fatalf("PoolSize = %d, want %d", opts.PoolSize, tt.wantPool)
+			}
+			client := redis.NewUniversalClient(opts)
+			defer client.Close()
+			if tt.wantCluster {
+				if _, ok := client.(*redis.ClusterClient); !ok {
+					t.Fatalf("client type = %T, want *redis.ClusterClient", client)
+				}
+			} else if _, ok := client.(*redis.Client); !ok {
+				t.Fatalf("client type = %T, want *redis.Client", client)
+			}
+		})
+	}
+}
+
+func TestNewRedisClientPingsConfiguredRedis(t *testing.T) {
+	redisURL := os.Getenv("REDIS_TEST_URL")
+	if redisURL == "" {
+		t.Skip("REDIS_TEST_URL is not configured")
+	}
+	client, err := NewRedisClient(RedisConfig{URL: redisURL})
+	if err != nil {
+		t.Fatalf("NewRedisClient() error = %v", err)
+	}
+	t.Cleanup(func() { _ = client.Close() })
+}
+
+func TestNewRedisOptionsRejectsInvalidClusterConfiguration(t *testing.T) {
+	tests := []struct {
+		name string
+		cfg  RedisConfig
+	}{
+		{
+			name: "multiple addresses without cluster mode",
+			cfg:  RedisConfig{URL: "redis://node-a:6379,node-b:6379/0"},
+		},
+		{
+			name: "nonzero database in cluster mode",
+			cfg:  RedisConfig{URL: "redis://node-a:6379/1", ClusterMode: true},
+		},
+		{
+			name: "empty cluster address",
+			cfg:  RedisConfig{URL: "redis://node-a:6379,,node-b:6379/0", ClusterMode: true},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if _, err := NewRedisOptions(tt.cfg); err == nil {
+				t.Fatal("NewRedisOptions() error = nil")
+			}
+		})
+	}
+}

server/internal/handler/invitation_rate_limiter.go

@@ -62,7 +62,7 @@ func NewMemoryInvitationRateLimiters(limits InvitationRateLimits) InvitationRate
 
 // NewRedisInvitationRateLimiters builds replica-shared gates with isolated key
 // namespaces for each dimension.
-func NewRedisInvitationRateLimiters(rdb *redis.Client, limits InvitationRateLimits) InvitationRateLimiters {
+func NewRedisInvitationRateLimiters(rdb redis.UniversalClient, limits InvitationRateLimits) InvitationRateLimiters {
 	return InvitationRateLimiters{
 		Actor:     newRedisSlidingWindowRateLimiter(rdb, limits.Actor, invitationActorLimiterKeyPrefix),
 		Workspace: newRedisSlidingWindowRateLimiter(rdb, limits.Workspace, invitationWorkspaceLimiterKeyPrefix),

server/internal/handler/runtime_liveness_store.go

@@ -77,10 +77,10 @@ func runtimeLivenessKey(runtimeID string) string {
 // of an unexpired key is the signal "this runtime is alive right now"; the
 // sweeper consults this before marking a stale-in-DB runtime offline.
 type RedisLivenessStore struct {
-	rdb *redis.Client
+	rdb redis.UniversalClient
 }
 
-func NewRedisLivenessStore(rdb *redis.Client) *RedisLivenessStore {
+func NewRedisLivenessStore(rdb redis.UniversalClient) *RedisLivenessStore {
 	return &RedisLivenessStore{rdb: rdb}
 }
 
@@ -107,15 +107,25 @@ func (s *RedisLivenessStore) IsAliveBatch(ctx context.Context, runtimeIDs []stri
 	for i, id := range runtimeIDs {
 		keys[i] = runtimeLivenessKey(id)
 	}
-	values, err := s.rdb.MGet(ctx, keys...).Result()
-	if err != nil {
+	pipe := s.rdb.Pipeline()
+	commands := make([]*redis.StringCmd, len(keys))
+	for i, key := range keys {
+		commands[i] = pipe.Get(ctx, key)
+	}
+	_, err := pipe.Exec(ctx)
+	if err != nil && !errors.Is(err, redis.Nil) {
 		slog.Warn("liveness mget failed; falling back to DB",
 			"error", err, "count", len(keys))
 		return nil, false
 	}
 	out := make(map[string]bool, len(runtimeIDs))
-	for i, id := range runtimeIDs {
-		out[id] = values[i] != nil
+	for i, command := range commands {
+		if err := command.Err(); err != nil && !errors.Is(err, redis.Nil) {
+			slog.Warn("liveness get failed; falling back to DB",
+				"error", err, "count", len(keys))
+			return nil, false
+		}
+		out[runtimeIDs[i]] = command.Err() == nil
 	}
 	return out, true
 }

server/internal/handler/runtime_local_skills_redis_store.go

@@ -72,10 +72,10 @@ func localSkillImportPendingKey(runtimeID string) string {
 // RedisLocalSkillListStore stores pending / running / completed list requests
 // in Redis so every API node agrees on the same state.
 type RedisLocalSkillListStore struct {
-	rdb *redis.Client
+	rdb redis.UniversalClient
 }
 
-func NewRedisLocalSkillListStore(rdb *redis.Client) *RedisLocalSkillListStore {
+func NewRedisLocalSkillListStore(rdb redis.UniversalClient) *RedisLocalSkillListStore {
 	return &RedisLocalSkillListStore{rdb: rdb}
 }
 
@@ -266,10 +266,10 @@ func (s *RedisLocalSkillListStore) Fail(ctx context.Context, id string, errMsg s
 // request shape carries import-specific fields (skill_key, optional rename,
 // creator id) and Go generics don't buy us much for two concrete impls.
 type RedisLocalSkillImportStore struct {
-	rdb *redis.Client
+	rdb redis.UniversalClient
 }
 
-func NewRedisLocalSkillImportStore(rdb *redis.Client) *RedisLocalSkillImportStore {
+func NewRedisLocalSkillImportStore(rdb redis.UniversalClient) *RedisLocalSkillImportStore {
 	return &RedisLocalSkillImportStore{rdb: rdb}
 }
 

server/internal/handler/runtime_model_catalog_redis_cache.go

@@ -29,11 +29,11 @@ func modelCatalogKey(runtimeID string) string { return modelCatalogKeyPrefix + r
 
 // RedisModelCatalogCache is the multi-node implementation of ModelCatalogCache.
 type RedisModelCatalogCache struct {
-	rdb *redis.Client
+	rdb redis.UniversalClient
 	ttl time.Duration
 }
 
-func NewRedisModelCatalogCache(rdb *redis.Client) *RedisModelCatalogCache {
+func NewRedisModelCatalogCache(rdb redis.UniversalClient) *RedisModelCatalogCache {
 	return &RedisModelCatalogCache{rdb: rdb, ttl: modelCatalogServeWindow}
 }
 

server/internal/handler/runtime_models_redis_store.go

@@ -42,10 +42,10 @@ func modelListPendingKey(runtimeID string) string { return modelListPendingPrefi
 // RedisModelListStore stores model list requests in Redis so every API node
 // agrees on the same pending / running / terminal state.
 type RedisModelListStore struct {
-	rdb *redis.Client
+	rdb redis.UniversalClient
 }
 
-func NewRedisModelListStore(rdb *redis.Client) *RedisModelListStore {
+func NewRedisModelListStore(rdb redis.UniversalClient) *RedisModelListStore {
 	return &RedisModelListStore{rdb: rdb}
 }
 

server/internal/handler/runtime_update_redis_store.go

@@ -36,10 +36,10 @@ return 0
 `)
 
 type RedisUpdateStore struct {
-	rdb *redis.Client
+	rdb redis.UniversalClient
 }
 
-func NewRedisUpdateStore(rdb *redis.Client) *RedisUpdateStore {
+func NewRedisUpdateStore(rdb redis.UniversalClient) *RedisUpdateStore {
 	return &RedisUpdateStore{rdb: rdb}
 }
 

server/internal/handler/webhook_rate_limiter.go

@@ -287,25 +287,25 @@ func webhookLimiterAllowSource() string { return webhookLimiterAllowSrc }
 
 type redisWebhookRateLimiter struct {
 	cfg       SlidingWindowRateLimit
-	rdb       *redis.Client
+	rdb       redis.UniversalClient
 	keyPrefix string
 }
 
-func newRedisSlidingWindowRateLimiter(rdb *redis.Client, cfg SlidingWindowRateLimit, keyPrefix string) SlidingWindowRateLimiter {
+func newRedisSlidingWindowRateLimiter(rdb redis.UniversalClient, cfg SlidingWindowRateLimit, keyPrefix string) SlidingWindowRateLimiter {
 	return &redisWebhookRateLimiter{cfg: cfg, rdb: rdb, keyPrefix: keyPrefix}
 }
 
-func NewRedisWebhookRateLimiter(rdb *redis.Client, cfg WebhookRateLimit) WebhookRateLimiter {
+func NewRedisWebhookRateLimiter(rdb redis.UniversalClient, cfg WebhookRateLimit) WebhookRateLimiter {
 	return newRedisSlidingWindowRateLimiter(rdb, cfg, webhookLimiterKeyPrefix)
 }
 
 // NewRedisWebhookIPRateLimiter is the per-IP variant: same sliding-window
 // Lua script, different key namespace so the two budgets don't interfere.
-func NewRedisWebhookIPRateLimiter(rdb *redis.Client, cfg WebhookRateLimit) WebhookRateLimiter {
+func NewRedisWebhookIPRateLimiter(rdb redis.UniversalClient, cfg WebhookRateLimit) WebhookRateLimiter {
 	return newRedisSlidingWindowRateLimiter(rdb, cfg, webhookIPLimiterKeyPrefix)
 }
 
-func NewRedisWebhookAbsoluteIPRateLimiter(rdb *redis.Client, cfg WebhookRateLimit) WebhookRateLimiter {
+func NewRedisWebhookAbsoluteIPRateLimiter(rdb redis.UniversalClient, cfg WebhookRateLimit) WebhookRateLimiter {
 	return newRedisSlidingWindowRateLimiter(rdb, cfg, webhookAbsoluteIPLimiterKeyPrefix)
 }
 

server/internal/integrations/channel/engine/redis_lease_store.go

@@ -45,11 +45,11 @@ var redisReleaseLease = redis.NewScript(redisReleaseLeaseSource)
 // client/pool. Every mutation is a single Lua operation, so compare + expiry
 // update/delete cannot be interleaved by another replica.
 type RedisLeaseStore struct {
-	client    *redis.Client
+	client    redis.UniversalClient
 	namespace string
 }
 
-func NewRedisLeaseStore(client *redis.Client, namespace string) (*RedisLeaseStore, error) {
+func NewRedisLeaseStore(client redis.UniversalClient, namespace string) (*RedisLeaseStore, error) {
 	if client == nil {
 		return nil, errors.New("channel lease redis client is nil")
 	}
@@ -108,15 +108,20 @@ func (s *RedisLeaseStore) ListHeldWSLeases(ctx context.Context, ids []pgtype.UUI
 	for i, id := range ids {
 		keys[i] = s.key(id)
 	}
-	// RedisLeaseStore intentionally takes *redis.Client (standalone/sentinel),
-	// not ClusterClient. A future cluster-mode implementation must group keys by
-	// hash slot or pipeline individual GETs; cross-slot MGET is not valid.
-	values, err := s.client.MGet(ctx, keys...).Result()
-	if err != nil {
+	pipe := s.client.Pipeline()
+	commands := make([]*redis.StringCmd, len(keys))
+	for i, key := range keys {
+		commands[i] = pipe.Get(ctx, key)
+	}
+	_, err := pipe.Exec(ctx)
+	if err != nil && !errors.Is(err, redis.Nil) {
 		return nil, err
 	}
-	for i, value := range values {
-		if value != nil {
+	for i, command := range commands {
+		if err := command.Err(); err != nil && !errors.Is(err, redis.Nil) {
+			return nil, err
+		}
+		if command.Err() == nil {
 			held[uuidString(ids[i])] = struct{}{}
 		}
 	}

server/internal/integrations/channel/engine/redis_lease_store_test.go

@@ -8,6 +8,7 @@ import (
 
 	redismock "github.com/go-redis/redismock/v9"
 	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/redis/go-redis/v9"
 )
 
 func TestRedisLeaseStoreRequiresSafeNamespace(t *testing.T) {
@@ -82,7 +83,8 @@ func TestRedisLeaseStoreListsOnlyHeldKeys(t *testing.T) {
 	}
 	id1 := uuidFromString(t, "13131313-1313-1313-1313-131313131313")
 	id2 := uuidFromString(t, "14141414-1414-1414-1414-141414141414")
-	mock.ExpectMGet(store.key(id1), store.key(id2)).SetVal([]interface{}{"owner", nil})
+	mock.ExpectGet(store.key(id1)).SetVal("owner")
+	mock.ExpectGet(store.key(id2)).SetErr(redis.Nil)
 	held, err := store.ListHeldWSLeases(context.Background(), []pgtype.UUID{id1, id2})
 	if err != nil {
 		t.Fatal(err)

server/internal/integrations/wecom/dedupe_redis.go

@@ -18,12 +18,13 @@ import (
 	"log/slog"
 	"time"
 
+	"github.com/multica-ai/multica/server/internal/database"
 	"github.com/redis/go-redis/v9"
 )
 
 // redisDedupe is a DedupeStore backed by one Redis key per delivery.
 type redisDedupe struct {
-	rdb    *redis.Client
+	rdb    redis.UniversalClient
 	log    *slog.Logger
 	budget time.Duration
 }
@@ -32,7 +33,7 @@ type redisDedupe struct {
 // which RelayOutbound reads as "no cross-restart claim" — correct only where
 // there is no relay either. A budget of zero takes defaultClaimBudget; tests
 // shrink it so the outcome grace derived from it stays a test's worth of time.
-func NewRedisDedupe(rdb *redis.Client, budget time.Duration, log *slog.Logger) DedupeStore {
+func NewRedisDedupe(rdb redis.UniversalClient, budget time.Duration, log *slog.Logger) DedupeStore {
 	if rdb == nil {
 		return nil
 	}
@@ -50,7 +51,7 @@ func NewRedisDedupe(rdb *redis.Client, budget time.Duration, log *slog.Logger) D
 	// exit budget it promised by however far the socket timeout reaches.
 	// Production passes a client built for this (cmd/server: newClaimRedisClient);
 	// anything else is a wiring mistake, and a silent one, so it is named here.
-	if !rdb.Options().ContextTimeoutEnabled {
+	if enabled, known := database.RedisContextTimeoutEnabled(rdb); known && !enabled {
 		log.Warn("wecom relay: claim store client ignores context deadlines; " +
 			"claim budgets and the shutdown drain budget will not be honoured " +
 			"(set redis.Options.ContextTimeoutEnabled)")

server/internal/middleware/plugin_ratelimit.go

@@ -15,7 +15,7 @@ import (
 // credential. The token is hashed before it is used as a Redis key; secrets
 // must never enter logs or operational data. A missing Redis client fails open,
 // matching the existing public-edge limiter behavior.
-func PluginRateLimit(rdb *redis.Client, limit int, window time.Duration) func(http.Handler) http.Handler {
+func PluginRateLimit(rdb redis.UniversalClient, limit int, window time.Duration) func(http.Handler) http.Handler {
 	return func(next http.Handler) http.Handler {
 		if rdb == nil {
 			return next

server/internal/middleware/ratelimit.go

@@ -57,7 +57,7 @@ func ParseTrustedProxies(raw string) []*net.IPNet {
 // When the list is empty (default), XFF is never consulted — only
 // RemoteAddr is used. This matches the project's conservative trust
 // model (see health_realtime.go).
-func RateLimit(rdb *redis.Client, limit int, window time.Duration, trustedProxies []*net.IPNet) func(http.Handler) http.Handler {
+func RateLimit(rdb redis.UniversalClient, limit int, window time.Duration, trustedProxies []*net.IPNet) func(http.Handler) http.Handler {
 	return func(next http.Handler) http.Handler {
 		if rdb == nil {
 			return next

server/internal/realtime/redis_relay.go

@@ -133,8 +133,8 @@ func deliverEnvelope(hub *Hub, daemonRuntime DaemonRuntimeDeliverer, wecomOutbou
 // subscribers. Local fanout is delegated to the wrapped *Hub.
 type RedisRelay struct {
 	hub       *Hub
-	writeRDB  *redis.Client
-	readRDB   *redis.Client
+	writeRDB  redis.UniversalClient
+	readRDB   redis.UniversalClient
 	nodeID    string
 	retention StreamRetentionConfig
 	ttl       *streamTTLRefresher
@@ -161,7 +161,7 @@ type scopeConsumer struct {
 
 // NewRedisRelay constructs a relay. The caller is responsible for invoking
 // Start before producing messages.
-func NewRedisRelay(hub *Hub, rdb *redis.Client) *RedisRelay {
+func NewRedisRelay(hub *Hub, rdb redis.UniversalClient) *RedisRelay {
 	return NewRedisRelayWithClients(hub, rdb, rdb)
 }
 
@@ -169,13 +169,13 @@ func NewRedisRelay(hub *Hub, rdb *redis.Client) *RedisRelay {
 // writes and blocking reads. The read client is reserved for XREADGROUP BLOCK
 // calls so long-polling stream consumers cannot exhaust the pool used by XADD,
 // heartbeats, acks, and other request-path Redis operations.
-func NewRedisRelayWithClients(hub *Hub, writeRDB, readRDB *redis.Client) *RedisRelay {
+func NewRedisRelayWithClients(hub *Hub, writeRDB, readRDB redis.UniversalClient) *RedisRelay {
 	return NewRedisRelayWithClientsAndConfig(hub, writeRDB, readRDB, DefaultStreamRetentionConfig())
 }
 
 // NewRedisRelayWithClientsAndConfig applies the same stream retention controls
 // used by sharded mode so legacy and dual-mode rollouts cannot silently diverge.
-func NewRedisRelayWithClientsAndConfig(hub *Hub, writeRDB, readRDB *redis.Client, retention StreamRetentionConfig) *RedisRelay {
+func NewRedisRelayWithClientsAndConfig(hub *Hub, writeRDB, readRDB redis.UniversalClient, retention StreamRetentionConfig) *RedisRelay {
 	if readRDB == nil {
 		readRDB = writeRDB
 	}

server/internal/realtime/sharded_stream_relay.go

@@ -155,8 +155,8 @@ func (c ShardedStreamRelayConfig) Validate() error {
 // bounded by pod_count * shard_count instead of active_scope_count.
 type ShardedStreamRelay struct {
 	hub      *Hub
-	writeRDB *redis.Client
-	readRDB  *redis.Client
+	writeRDB redis.UniversalClient
+	readRDB  redis.UniversalClient
 	nodeID   string
 	config   ShardedStreamRelayConfig
 	now      func() time.Time
@@ -173,7 +173,7 @@ type ShardedStreamRelay struct {
 	wecomOutbound WecomOutboundDeliverer
 }
 
-func NewShardedStreamRelay(hub *Hub, writeRDB, readRDB *redis.Client, config ShardedStreamRelayConfig) *ShardedStreamRelay {
+func NewShardedStreamRelay(hub *Hub, writeRDB, readRDB redis.UniversalClient, config ShardedStreamRelayConfig) *ShardedStreamRelay {
 	if readRDB == nil {
 		readRDB = writeRDB
 	}

server/internal/realtime/stream_retention.go

@@ -84,7 +84,7 @@ func newStreamTTLRefresher(ttl, refreshEvery time.Duration) *streamTTLRefresher
 	}
 }
 
-func (r *streamTTLRefresher) refreshIfDue(ctx context.Context, client *redis.Client, key string) error {
+func (r *streamTTLRefresher) refreshIfDue(ctx context.Context, client redis.UniversalClient, key string) error {
 	now := r.now()
 	if !r.claimRefresh(key, now) {
 		return nil
@@ -104,7 +104,7 @@ func (r *streamTTLRefresher) refreshIfDue(ctx context.Context, client *redis.Cli
 
 // repairMissingTTL assigns a TTL only when a stream exists without one. It
 // intentionally does not refresh a healthy TTL, so an idle stream can expire.
-func (r *streamTTLRefresher) repairMissingTTL(ctx context.Context, client *redis.Client, key string) (time.Duration, error) {
+func (r *streamTTLRefresher) repairMissingTTL(ctx context.Context, client redis.UniversalClient, key string) (time.Duration, error) {
 	return r.reconcileTTL(ctx, client, key, true)
 }
 
@@ -112,7 +112,7 @@ func (r *streamTTLRefresher) repairMissingTTL(ctx context.Context, client *redis
 // TTL when disabled. The disabled path is the compatibility and rollback
 // phase: once all new replicas have observed it, old binaries can keep writing
 // without an inherited expiry deleting an active stream.
-func (r *streamTTLRefresher) reconcileTTL(ctx context.Context, client *redis.Client, key string, enabled bool) (time.Duration, error) {
+func (r *streamTTLRefresher) reconcileTTL(ctx context.Context, client redis.UniversalClient, key string, enabled bool) (time.Duration, error) {
 	ttl, err := client.PTTL(ctx, key).Result()
 	if err != nil {
 		return 0, err

server/internal/service/empty_claim_cache.go

@@ -17,15 +17,15 @@ import (
 // which closes the race where a slow claim writes an empty verdict
 // AFTER an enqueue has already invalidated it:
 //
-//   T1 claim:   v0 := GET version
-//               SELECT ... -> empty
-//               (slow, e.g. GC pause)
-//   T2 enqueue: INSERT row
-//               INCR version  (-> v1)
-//               wakeup
-//   T1 claim:   SET empty = v0
-//   T3 claim:   v1' := GET version (== v1)
-//               GET empty (== v0) -> v0 != v1, treat as miss -> SELECT
+//	T1 claim:   v0 := GET version
+//	            SELECT ... -> empty
+//	            (slow, e.g. GC pause)
+//	T2 enqueue: INSERT row
+//	            INCR version  (-> v1)
+//	            wakeup
+//	T1 claim:   SET empty = v0
+//	T3 claim:   v1' := GET version (== v1)
+//	            GET empty (== v0) -> v0 != v1, treat as miss -> SELECT
 //
 // Without the version tag T3 would have hit the stale empty key and
 // the just-queued task would sit idle until the empty key's TTL
@@ -69,20 +69,20 @@ const emptyClaimRedisTimeout = 250 * time.Millisecond
 // single-node dev / tests with no REDIS_URL degrade cleanly to direct
 // DB lookups.
 type EmptyClaimCache struct {
-	rdb *redis.Client
+	rdb redis.UniversalClient
 }
 
 // NewEmptyClaimCache returns a cache backed by rdb. Pass nil to
 // disable caching; the returned *EmptyClaimCache is safe to call but
 // never hits Redis.
-func NewEmptyClaimCache(rdb *redis.Client) *EmptyClaimCache {
+func NewEmptyClaimCache(rdb redis.UniversalClient) *EmptyClaimCache {
 	if rdb == nil {
 		return nil
 	}
 	return &EmptyClaimCache{rdb: rdb}
 }
 
-func emptyClaimKey(runtimeID string) string   { return emptyClaimCachePrefix + runtimeID }
+func emptyClaimKey(runtimeID string) string     { return emptyClaimCachePrefix + runtimeID }
 func emptyClaimVersion(runtimeID string) string { return emptyClaimVersionPrefix + runtimeID }
 
 func (c *EmptyClaimCache) bounded(ctx context.Context) (context.Context, context.CancelFunc) {
@@ -127,17 +127,24 @@ func (c *EmptyClaimCache) IsEmpty(ctx context.Context, runtimeID string) bool {
 	}
 	bctx, cancel := c.bounded(ctx)
 	defer cancel()
-	// MGET returns []interface{} of either the value (string) or nil.
-	vals, err := c.rdb.MGet(bctx, emptyClaimKey(runtimeID), emptyClaimVersion(runtimeID)).Result()
+	// Keep the established key names so old and new replicas exchange the same
+	// invalidation signal during rolling deploys. They land in different Redis
+	// Cluster slots, so issue two GETs in one pipeline instead of using MGET.
+	// An atomic snapshot is unnecessary: versions only increase, and enqueue
+	// completes Bump before waking the daemon. A claim triggered by that wakeup
+	// therefore observes the new version regardless of GET execution order.
+	pipe := c.rdb.Pipeline()
+	emptyCmd := pipe.Get(bctx, emptyClaimKey(runtimeID))
+	versionCmd := pipe.Get(bctx, emptyClaimVersion(runtimeID))
+	if _, err := pipe.Exec(bctx); err != nil && !errors.Is(err, redis.Nil) {
+		slog.Warn("empty_claim_cache: pipelined get failed; falling back to DB", "error", err)
+		return false
+	}
+	emptyVer, err := emptyCmd.Result()
 	if err != nil {
-		slog.Warn("empty_claim_cache: mget failed; falling back to DB", "error", err)
-		return false
-	}
-	if len(vals) != 2 || vals[0] == nil {
-		return false
-	}
-	emptyVer, ok := vals[0].(string)
-	if !ok {
+		if !errors.Is(err, redis.Nil) {
+			slog.Warn("empty_claim_cache: empty verdict get failed; falling back to DB", "error", err)
+		}
 		return false
 	}
 	// A missing version key means "no enqueue has ever bumped this
@@ -146,10 +153,11 @@ func (c *EmptyClaimCache) IsEmpty(ctx context.Context, runtimeID string) bool {
 	// must match here, otherwise the fast path would never trigger
 	// for fresh runtimes.
 	curVer := "0"
-	if vals[1] != nil {
-		if s, ok := vals[1].(string); ok {
-			curVer = s
-		}
+	if version, err := versionCmd.Result(); err == nil {
+		curVer = version
+	} else if !errors.Is(err, redis.Nil) {
+		slog.Warn("empty_claim_cache: version get failed; falling back to DB", "error", err)
+		return false
 	}
 	return emptyVer == curVer
 }

server/internal/service/empty_claim_cache_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	redismock "github.com/go-redis/redismock/v9"
 	"github.com/redis/go-redis/v9"
 )
 
@@ -40,6 +41,31 @@ func newRedisTestClient(t *testing.T) *redis.Client {
 	return rdb
 }
 
+func TestEmptyClaimKeysRemainRollingDeploymentCompatible(t *testing.T) {
+	const runtimeID = "runtime-a"
+	if got, want := emptyClaimKey(runtimeID), "mul:claim:runtime:empty:runtime-a"; got != want {
+		t.Fatalf("empty claim key = %q, want %q", got, want)
+	}
+	if got, want := emptyClaimVersion(runtimeID), "mul:claim:runtime:version:runtime-a"; got != want {
+		t.Fatalf("empty claim version key = %q, want %q", got, want)
+	}
+}
+
+func TestEmptyClaimCache_IsEmptyUsesClusterSafePipelinedGets(t *testing.T) {
+	rdb, mock := redismock.NewClientMock()
+	t.Cleanup(func() { _ = rdb.Close() })
+	c := NewEmptyClaimCache(rdb)
+
+	mock.ExpectGet(emptyClaimKey("runtime-a")).SetVal("7")
+	mock.ExpectGet(emptyClaimVersion("runtime-a")).SetVal("7")
+	if !c.IsEmpty(context.Background(), "runtime-a") {
+		t.Fatal("expected matching pipelined GET results to hit the cache")
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatal(err)
+	}
+}
+
 func TestEmptyClaimCache_NilSafe(t *testing.T) {
 	var c *EmptyClaimCache // nil
 	ctx := context.Background()

server/internal/service/reclaim_check_cache.go

@@ -2,6 +2,7 @@ package service
 
 import (
 	"context"
+	"errors"
 	"log/slog"
 	"strconv"
 	"time"
@@ -64,10 +65,10 @@ return #members
 // Redis errors always mean "check PostgreSQL now", so the cache can only reduce
 // load; it is never authoritative for task recovery.
 type ReclaimCheckCache struct {
-	rdb *redis.Client
+	rdb redis.UniversalClient
 }
 
-func NewReclaimCheckCache(rdb *redis.Client) *ReclaimCheckCache {
+func NewReclaimCheckCache(rdb redis.UniversalClient) *ReclaimCheckCache {
 	if rdb == nil {
 		return nil
 	}
@@ -118,14 +119,29 @@ func (c *ReclaimCheckCache) DueRuntimeIDs(ctx context.Context, runtimeIDs []stri
 	for _, runtimeID := range validRuntimeIDs {
 		backstopKeys = append(backstopKeys, reclaimCheckBackstopKey(runtimeID))
 	}
-	backstops, err := c.rdb.MGet(bctx, backstopKeys...).Result()
-	if err != nil {
+	backstopPipe := c.rdb.Pipeline()
+	backstopCommands := make([]*redis.StringCmd, len(backstopKeys))
+	for i, key := range backstopKeys {
+		backstopCommands[i] = backstopPipe.Get(bctx, key)
+	}
+	_, err := backstopPipe.Exec(bctx)
+	if err != nil && !errors.Is(err, redis.Nil) {
 		slog.Warn("reclaim_check_cache: backstop read failed; falling back to DB", "error", err)
 		return validRuntimeIDs
 	}
-	if len(backstops) != len(validRuntimeIDs) {
-		slog.Warn("reclaim_check_cache: incomplete backstop read; falling back to DB")
-		return validRuntimeIDs
+	backstops := make([]string, len(backstopCommands))
+	backstopExists := make([]bool, len(backstopCommands))
+	for i, command := range backstopCommands {
+		value, err := command.Result()
+		if errors.Is(err, redis.Nil) {
+			continue
+		}
+		if err != nil {
+			slog.Warn("reclaim_check_cache: backstop result failed; falling back to DB", "error", err)
+			return validRuntimeIDs
+		}
+		backstops[i] = value
+		backstopExists[i] = true
 	}
 
 	nowMillis := now.UnixMilli()
@@ -137,14 +153,13 @@ func (c *ReclaimCheckCache) DueRuntimeIDs(ctx context.Context, runtimeIDs []stri
 	pipe := c.rdb.Pipeline()
 	scheduleCommands := make([]scheduleCommand, 0, len(validRuntimeIDs))
 	for i, runtimeID := range validRuntimeIDs {
-		backstop, ok := backstops[i].(string)
-		if !ok {
+		if !backstopExists[i] {
 			// A missing or non-string value means this runtime has no trustworthy
 			// successful-check marker.
 			due[i] = true
 			continue
 		}
-		checkedMillis, err := strconv.ParseInt(backstop, 10, 64)
+		checkedMillis, err := strconv.ParseInt(backstops[i], 10, 64)
 		if err != nil {
 			due[i] = true
 			continue