// Package router 维护 room-service 的房间 owner 路由目录和 lease 语义。 package router import ( "context" "encoding/json" "strconv" "time" "github.com/redis/go-redis/v9" "hyapp/pkg/idgen" ) // roomRouteKeyPrefix 是生产房间 owner lease 的 Redis key 前缀。 const roomRouteKeyPrefix = "room:route:" // healthLeaseKey 是 ready 探针专用 key,避免探针污染真实房间路由。 const healthLeaseKey = "room:healthcheck:lease" const ensureOwnerScript = ` local current = redis.call("GET", KEYS[1]) if not current then redis.call("SET", KEYS[1], ARGV[3], "PX", ARGV[2]) return ARGV[3] end local ok, decoded = pcall(cjson.decode, current) if not ok then redis.call("SET", KEYS[1], ARGV[3], "PX", ARGV[2]) return ARGV[3] end local expires_at_ms = tonumber(decoded["expires_at_ms"]) or 0 if decoded["node_id"] == ARGV[4] and expires_at_ms > tonumber(ARGV[1]) then local next_ok, next_decoded = pcall(cjson.decode, ARGV[3]) if not next_ok then redis.call("SET", KEYS[1], ARGV[3], "PX", ARGV[2]) return ARGV[3] end decoded["room_id"] = next_decoded["room_id"] decoded["expires_at_ms"] = next_decoded["expires_at_ms"] local refreshed = cjson.encode(decoded) redis.call("SET", KEYS[1], refreshed, "PX", ARGV[2]) return refreshed end if expires_at_ms <= tonumber(ARGV[1]) then redis.call("SET", KEYS[1], ARGV[3], "PX", ARGV[2]) return ARGV[3] end return current ` const verifyOwnerScript = ` local current = redis.call("GET", KEYS[1]) if not current then return 0 end local ok, decoded = pcall(cjson.decode, current) if not ok then return 0 end if decoded["node_id"] == ARGV[2] and decoded["lease_token"] == ARGV[3] and tonumber(decoded["expires_at_ms"]) > tonumber(ARGV[1]) then return 1 end return 0 ` // RedisDirectory 用 Redis 保存 room_id 到执行节点的带 TTL 租约。 type RedisDirectory struct { // client 是 go-redis 客户端,连接生命周期由 app.App 关闭。 client *redis.Client } // redisLeaseValue 是 Redis JSON 载荷,字段名稳定便于排查和跨语言工具读取。 type redisLeaseValue struct { RoomID string `json:"room_id"` NodeID string `json:"node_id"` LeaseToken string `json:"lease_token"` ExpiresAt int64 `json:"expires_at_ms"` } // NewRedisDirectory 创建 Redis 目录实现。 func NewRedisDirectory(client *redis.Client) *RedisDirectory { return &RedisDirectory{client: client} } // NewRedisClient 创建 Redis 客户端,并验证连接可用。 func NewRedisClient(ctx context.Context, addr string, password string, db int) (*redis.Client, error) { // Redis 只保存热路由/lease,不作为房间恢复的唯一来源。 client := redis.NewClient(&redis.Options{ Addr: addr, Password: password, DB: db, }) if err := client.Ping(ctx).Err(); err != nil { // 启动探测失败时关闭客户端,避免调用方遗留半初始化连接。 _ = client.Close() return nil, err } return client, nil } // Ping 验证 Redis 当前可访问,调用方负责传入短 deadline。 func (d *RedisDirectory) Ping(ctx context.Context) error { // ready 探针调用该方法确认 Redis 基础连通性。 return d.client.Ping(ctx).Err() } // ProbeLease 使用与房间归属相同的 Lua 路径探测 Redis lease 原子操作能力。 func (d *RedisDirectory) ProbeLease(ctx context.Context, nodeID string, now time.Time, ttl time.Duration) error { // 探针必须走同一段 Lua,才能发现 Eval 权限或 cjson 支持问题。 next := Lease{ RoomID: "__healthcheck__", NodeID: nodeID, LeaseToken: idgen.New("lease"), ExpiresAt: now.Add(ttl), } payload, err := encodeLease(next) if err != nil { return err } // Lua 脚本保证 read-check-set 原子化,避免 GET + SET 竞争导致双 owner。 value, err := d.client.Eval(ctx, ensureOwnerScript, []string{healthLeaseKey}, strconv.FormatInt(now.UnixMilli(), 10), strconv.FormatInt(ttl.Milliseconds(), 10), payload, nodeID, ).Text() if err != nil { return err } // 返回值可能是已有租约或新租约,能解码即可证明脚本路径正常。 _, err = decodeLease(value) return err } // Lookup 查询房间当前 owner 租约。 func (d *RedisDirectory) Lookup(ctx context.Context, roomID string) (Lease, bool, error) { value, err := d.client.Get(ctx, routeKey(roomID)).Result() if err != nil { if err == redis.Nil { // key 不存在表示没有当前 owner。 return Lease{}, false, nil } return Lease{}, false, err } lease, err := decodeLease(value) if err != nil { // Redis 值损坏时不能静默接管,否则可能破坏单活。 return Lease{}, false, err } return lease, true, nil } // EnsureOwner 在租约不存在、过期或归当前节点时授予本节点执行权。 func (d *RedisDirectory) EnsureOwner(ctx context.Context, roomID string, nodeID string, now time.Time, ttl time.Duration) (Lease, error) { // next 是当前节点希望写入的租约;Lua 会决定是否真正采用它。 // 同节点续租时 Lua 会保留现有 lease_token,只刷新 expires_at_ms, // 避免后台 worker 续租把正在执行的同节点命令误判为旧 owner。 next := Lease{ RoomID: roomID, NodeID: nodeID, LeaseToken: idgen.New("lease"), ExpiresAt: now.Add(ttl), } payload, err := encodeLease(next) if err != nil { return Lease{}, err } // Redis TTL 和 payload.expires_at_ms 同时写入,TTL 用于自动清理,payload 用于脚本判断。 value, err := d.client.Eval(ctx, ensureOwnerScript, []string{routeKey(roomID)}, strconv.FormatInt(now.UnixMilli(), 10), strconv.FormatInt(ttl.Milliseconds(), 10), payload, nodeID, ).Text() if err != nil { return Lease{}, err } lease, err := decodeLease(value) if err != nil { // 返回值无法解码时调用方不能确认 owner,必须失败。 return Lease{}, err } return lease, nil } // VerifyOwner 校验调用方持有的 lease token 仍是当前有效 owner。 func (d *RedisDirectory) VerifyOwner(ctx context.Context, roomID string, nodeID string, leaseToken string, now time.Time) (bool, error) { value, err := d.client.Eval(ctx, verifyOwnerScript, []string{routeKey(roomID)}, strconv.FormatInt(now.UnixMilli(), 10), nodeID, leaseToken, ).Int() if err != nil { return false, err } return value == 1, nil } // ForceExpire 只服务测试,生产启动路径不会调用。 func (d *RedisDirectory) ForceExpire(roomID string) { // Redis 实现直接删除 key 来模拟 lease 过期,避免等待真实 TTL。 _ = d.client.Del(context.Background(), routeKey(roomID)).Err() } func routeKey(roomID string) string { // 路由 key 只由房间 ID 决定,所有节点必须计算出同一个 Redis key。 return roomRouteKeyPrefix + roomID } func encodeLease(lease Lease) (string, error) { // 使用 JSON 而不是二进制,方便线上通过 redis-cli 直接排查 owner。 payload, err := json.Marshal(redisLeaseValue{ RoomID: lease.RoomID, NodeID: lease.NodeID, LeaseToken: lease.LeaseToken, ExpiresAt: lease.ExpiresAt.UnixMilli(), }) if err != nil { return "", err } return string(payload), nil } func decodeLease(value string) (Lease, error) { var payload redisLeaseValue if err := json.Unmarshal([]byte(value), &payload); err != nil { // 损坏值直接返回错误,让上层 ready 或命令路径暴露问题。 return Lease{}, err } // Redis 中保存毫秒时间戳,恢复为 time.Time 便于 ValidAt 判断。 return Lease{ RoomID: payload.RoomID, NodeID: payload.NodeID, LeaseToken: payload.LeaseToken, ExpiresAt: time.UnixMilli(payload.ExpiresAt), }, nil }