diff --git a/.env.example b/.env.example index 7a02f152e..5dd22e19e 100644 --- a/.env.example +++ b/.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://:mypassword@127.0.0.1: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://:mypassword@xxx.serverless.use1.cache.amazonaws.com:6379/0 +# REDIS_CLUSTER_MODE=true +# +# 3. Native multi-node Redis Cluster +# Separate node addresses between @ and / with commas. +# REDIS_URL=redis://:mypassword@10.0.0.1: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 diff --git a/apps/docs/content/docs/environment-variables.ja.mdx b/apps/docs/content/docs/environment-variables.ja.mdx index 7352fe6e7..ea6b7475e 100644 --- a/apps/docs/content/docs/environment-variables.ja.mdx +++ b/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 あたり毎分の回数 | diff --git a/apps/docs/content/docs/environment-variables.ko.mdx b/apps/docs/content/docs/environment-variables.ko.mdx index c0725a29c..6d8d12621 100644 --- a/apps/docs/content/docs/environment-variables.ko.mdx +++ b/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분에 인증 코드 검증 허용 횟수 | diff --git a/apps/docs/content/docs/environment-variables.mdx b/apps/docs/content/docs/environment-variables.mdx index 6c5b6c893..213fec757 100644 --- a/apps/docs/content/docs/environment-variables.mdx +++ b/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 | diff --git a/apps/docs/content/docs/environment-variables.zh.mdx b/apps/docs/content/docs/environment-variables.zh.mdx index 0e1650aa2..574481473 100644 --- a/apps/docs/content/docs/environment-variables.zh.mdx +++ b/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 每分钟验证验证码的次数 | diff --git a/docker-compose.selfhost.yml b/docker-compose.selfhost.yml index 98a7311b1..ce64c1f90 100644 --- a/docker-compose.selfhost.yml +++ b/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'} diff --git a/server/cmd/server/main.go b/server/cmd/server/main.go index 8670b727d..7634ecb72 100644 --- a/server/cmd/server/main.go +++ b/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") diff --git a/server/cmd/server/main_test.go b/server/cmd/server/main_test.go index 1b120dc76..0bba5a89e 100644 --- a/server/cmd/server/main_test.go +++ b/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") diff --git a/server/cmd/server/router.go b/server/cmd/server/router.go index e0abcd3d7..e1dcdb141 100644 --- a/server/cmd/server/router.go +++ b/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 diff --git a/server/cmd/server/wecom_claim_redis_wiring_test.go b/server/cmd/server/wecom_claim_redis_wiring_test.go index 9dede4580..b5ba08ad7 100644 --- a/server/cmd/server/wecom_claim_redis_wiring_test.go +++ b/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 { diff --git a/server/internal/auth/cloud_pat.go b/server/internal/auth/cloud_pat.go index 596b73dad..fa9fa862a 100644 --- a/server/internal/auth/cloud_pat.go +++ b/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 diff --git a/server/internal/auth/daemon_token_cache.go b/server/internal/auth/daemon_token_cache.go index 31b3c3820..c79eca728 100644 --- a/server/internal/auth/daemon_token_cache.go +++ b/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 } diff --git a/server/internal/auth/membership_cache.go b/server/internal/auth/membership_cache.go index a23330c71..ace7bc77e 100644 --- a/server/internal/auth/membership_cache.go +++ b/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 } diff --git a/server/internal/auth/pat_cache.go b/server/internal/auth/pat_cache.go index 15652e1fc..e161aba26 100644 --- a/server/internal/auth/pat_cache.go +++ b/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 } diff --git a/server/internal/database/redis.go b/server/internal/database/redis.go new file mode 100644 index 000000000..02b375d10 --- /dev/null +++ b/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, + } +} diff --git a/server/internal/database/redis_test.go b/server/internal/database/redis_test.go new file mode 100644 index 000000000..f64c1a0e4 --- /dev/null +++ b/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:secret@127.0.0.1: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://:secret@cache.example.com: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://:secret@10.0.0.1: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") + } + }) + } +} diff --git a/server/internal/handler/invitation_rate_limiter.go b/server/internal/handler/invitation_rate_limiter.go index 91bc5b317..b8bf3ae6c 100644 --- a/server/internal/handler/invitation_rate_limiter.go +++ b/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), diff --git a/server/internal/handler/runtime_liveness_store.go b/server/internal/handler/runtime_liveness_store.go index c14a39f82..cfab0d125 100644 --- a/server/internal/handler/runtime_liveness_store.go +++ b/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 } diff --git a/server/internal/handler/runtime_local_skills_redis_store.go b/server/internal/handler/runtime_local_skills_redis_store.go index 28345a998..483d5c67c 100644 --- a/server/internal/handler/runtime_local_skills_redis_store.go +++ b/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} } diff --git a/server/internal/handler/runtime_model_catalog_redis_cache.go b/server/internal/handler/runtime_model_catalog_redis_cache.go index 60f27d133..e5bc97284 100644 --- a/server/internal/handler/runtime_model_catalog_redis_cache.go +++ b/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} } diff --git a/server/internal/handler/runtime_models_redis_store.go b/server/internal/handler/runtime_models_redis_store.go index 23ad881c1..cf200d2ee 100644 --- a/server/internal/handler/runtime_models_redis_store.go +++ b/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} } diff --git a/server/internal/handler/runtime_update_redis_store.go b/server/internal/handler/runtime_update_redis_store.go index 5cdeab056..d6a630f55 100644 --- a/server/internal/handler/runtime_update_redis_store.go +++ b/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} } diff --git a/server/internal/handler/webhook_rate_limiter.go b/server/internal/handler/webhook_rate_limiter.go index 2d0de3ddc..5025b126c 100644 --- a/server/internal/handler/webhook_rate_limiter.go +++ b/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) } diff --git a/server/internal/integrations/channel/engine/redis_lease_store.go b/server/internal/integrations/channel/engine/redis_lease_store.go index 6e5fefcc8..10391f610 100644 --- a/server/internal/integrations/channel/engine/redis_lease_store.go +++ b/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{}{} } } diff --git a/server/internal/integrations/channel/engine/redis_lease_store_test.go b/server/internal/integrations/channel/engine/redis_lease_store_test.go index 6e799b22c..5234338b4 100644 --- a/server/internal/integrations/channel/engine/redis_lease_store_test.go +++ b/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) diff --git a/server/internal/integrations/wecom/dedupe_redis.go b/server/internal/integrations/wecom/dedupe_redis.go index 803358f80..c9f7ac899 100644 --- a/server/internal/integrations/wecom/dedupe_redis.go +++ b/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)") diff --git a/server/internal/middleware/plugin_ratelimit.go b/server/internal/middleware/plugin_ratelimit.go index c3a1b855e..470236efc 100644 --- a/server/internal/middleware/plugin_ratelimit.go +++ b/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 diff --git a/server/internal/middleware/ratelimit.go b/server/internal/middleware/ratelimit.go index 6fba26537..5f014fbac 100644 --- a/server/internal/middleware/ratelimit.go +++ b/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 diff --git a/server/internal/realtime/redis_relay.go b/server/internal/realtime/redis_relay.go index 5800ad8fa..649c29570 100644 --- a/server/internal/realtime/redis_relay.go +++ b/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 } diff --git a/server/internal/realtime/sharded_stream_relay.go b/server/internal/realtime/sharded_stream_relay.go index 7edd1aad5..9ba139e9b 100644 --- a/server/internal/realtime/sharded_stream_relay.go +++ b/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 } diff --git a/server/internal/realtime/stream_retention.go b/server/internal/realtime/stream_retention.go index 3409661ad..32f2199cd 100644 --- a/server/internal/realtime/stream_retention.go +++ b/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 diff --git a/server/internal/service/empty_claim_cache.go b/server/internal/service/empty_claim_cache.go index ca7872334..050a3d592 100644 --- a/server/internal/service/empty_claim_cache.go +++ b/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 } diff --git a/server/internal/service/empty_claim_cache_test.go b/server/internal/service/empty_claim_cache_test.go index 8d7b936bb..3ea66fdda 100644 --- a/server/internal/service/empty_claim_cache_test.go +++ b/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() diff --git a/server/internal/service/reclaim_check_cache.go b/server/internal/service/reclaim_check_cache.go index b0004032f..38735ffdd 100644 --- a/server/internal/service/reclaim_check_cache.go +++ b/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