213 lines
7.3 KiB
Go
213 lines
7.3 KiB
Go
// Package router 维护 room-service 的房间 owner 路由目录和 lease 语义。
|
||
package router
|
||
|
||
import (
|
||
"context"
|
||
"sync"
|
||
"time"
|
||
|
||
"hyapp/pkg/idgen"
|
||
)
|
||
|
||
// Lease 表达房间当前执行权归属。
|
||
type Lease struct {
|
||
// RoomID 是租约所属房间。
|
||
RoomID string
|
||
// NodeID 是当前拥有房间执行权的 room-service 实例。
|
||
NodeID string
|
||
// LeaseToken 是租约实例标识，便于未来做续约和 fencing。
|
||
LeaseToken string
|
||
// ExpiresAt 是租约过期时间，过期后其他节点可以接管并恢复房间。
|
||
ExpiresAt time.Time
|
||
}
|
||
|
||
// NodeRegistration 记录一个 room-service 实例可被其他实例直连的内网地址。
|
||
type NodeRegistration struct {
|
||
// NodeID 必须和 Redis lease 中的 owner node_id 完全一致。
|
||
NodeID string
|
||
// GRPCAddr 是实例私网 gRPC 地址，必须指向单个实例，不能写 room-service 负载均衡地址。
|
||
GRPCAddr string
|
||
// UpdatedAt 是最近一次心跳写入时间，便于排查节点注册是否卡住。
|
||
UpdatedAt time.Time
|
||
// ExpiresAt 是节点注册过期时间；过期节点不能作为 owner 转发目标。
|
||
ExpiresAt time.Time
|
||
}
|
||
|
||
// ValidAt 判断节点注册在指定时间是否仍可作为转发目标。
|
||
func (n NodeRegistration) ValidAt(now time.Time) bool {
|
||
// 没有地址或过期时间的注册不能被使用，避免把请求转发到空地址或旧实例。
|
||
return n.NodeID != "" && n.GRPCAddr != "" && !n.ExpiresAt.IsZero() && now.Before(n.ExpiresAt)
|
||
}
|
||
|
||
// ValidAt 判断租约在指定时刻是否仍有效。
|
||
func (l Lease) ValidAt(now time.Time) bool {
|
||
// 零值过期时间永远无效，避免空 lease 被误认为可继续持有。
|
||
return !l.ExpiresAt.IsZero() && now.Before(l.ExpiresAt)
|
||
}
|
||
|
||
// Directory 抽象 room_id -> node_id 的房间归属目录。
|
||
type Directory interface {
|
||
// Lookup 查询当前记录的房间 owner，不会主动接管。
|
||
Lookup(ctx context.Context, roomID string) (Lease, bool, error)
|
||
// EnsureOwner 在无 owner、owner 过期或 owner 是当前节点时授予当前节点执行权。
|
||
EnsureOwner(ctx context.Context, roomID string, nodeID string, now time.Time, ttl time.Duration) (Lease, error)
|
||
// VerifyOwner 校验当前节点和 lease token 仍然持有执行权，不续租也不接管。
|
||
VerifyOwner(ctx context.Context, roomID string, nodeID string, leaseToken string, now time.Time) (bool, error)
|
||
// ReleaseOwner 在节点安全下线后释放自己仍持有的 owner lease，让新节点无需等待 TTL 才能恢复。
|
||
ReleaseOwner(ctx context.Context, roomID string, nodeID string) (bool, error)
|
||
// ForceExpire 只用于测试模拟租约过期。
|
||
ForceExpire(roomID string)
|
||
}
|
||
|
||
// NodeRegistry 维护 room-service node_id 到实例私网 gRPC 地址的短租约。
|
||
type NodeRegistry interface {
|
||
// RegisterNode 写入或续约当前节点地址；ttl 过期后其他节点不能继续转发到该实例。
|
||
RegisterNode(ctx context.Context, node NodeRegistration, ttl time.Duration) error
|
||
// LookupNode 查询一个 owner node_id 的实例地址。
|
||
LookupNode(ctx context.Context, nodeID string) (NodeRegistration, bool, error)
|
||
// UnregisterNode 在实例安全退出后删除注册，避免转发到已停止进程。
|
||
UnregisterNode(ctx context.Context, nodeID string) error
|
||
}
|
||
|
||
// MemoryDirectory 是首版测试与本地运行使用的内存目录实现。
|
||
type MemoryDirectory struct {
|
||
// mu 保护 leases，测试里可能用多个服务实例模拟接管。
|
||
mu sync.Mutex
|
||
// leases 保存 room_id -> lease。
|
||
leases map[string]Lease
|
||
// nodes 保存 node_id -> 注册地址，供 transport 层 owner 转发测试复用。
|
||
nodes map[string]NodeRegistration
|
||
}
|
||
|
||
// NewMemoryDirectory 初始化空的内存目录。
|
||
func NewMemoryDirectory() *MemoryDirectory {
|
||
return &MemoryDirectory{
|
||
leases: make(map[string]Lease),
|
||
nodes: make(map[string]NodeRegistration),
|
||
}
|
||
}
|
||
|
||
// Lookup 查询当前房间归属。
|
||
func (d *MemoryDirectory) Lookup(_ context.Context, roomID string) (Lease, bool, error) {
|
||
d.mu.Lock()
|
||
defer d.mu.Unlock()
|
||
|
||
// 内存实现不主动清理过期 lease，EnsureOwner 会在写路径判断有效性。
|
||
lease, ok := d.leases[roomID]
|
||
|
||
return lease, ok, nil
|
||
}
|
||
|
||
// EnsureOwner 在租约不存在、已过期或已归当前节点时，授予本节点执行权。
|
||
func (d *MemoryDirectory) EnsureOwner(_ context.Context, roomID string, nodeID string, now time.Time, ttl time.Duration) (Lease, error) {
|
||
d.mu.Lock()
|
||
defer d.mu.Unlock()
|
||
|
||
current, ok := d.leases[roomID]
|
||
if ok && current.ValidAt(now) && current.NodeID != nodeID {
|
||
// 其他节点持有有效 lease 时返回现有 owner，调用方应拒绝执行房间命令。
|
||
return current, nil
|
||
}
|
||
|
||
if ok && current.ValidAt(now) && current.NodeID == nodeID {
|
||
// 同一节点续租只能延长过期时间，不能轮换 token。
|
||
// Room Cell 命令在进入 mailbox 前拿到 lease，提交前会用 token 做 fencing；
|
||
// 如果后台 sweep 续租时换 token，就会把同节点正在执行的命令误判为旧 owner。
|
||
current.ExpiresAt = now.Add(ttl)
|
||
current.RoomID = roomID
|
||
d.leases[roomID] = current
|
||
return current, nil
|
||
}
|
||
|
||
// 没有 lease 或 lease 已过期时生成新 token；这代表新的执行权实例。
|
||
next := Lease{
|
||
RoomID: roomID,
|
||
NodeID: nodeID,
|
||
LeaseToken: idgen.New("lease"),
|
||
ExpiresAt: now.Add(ttl),
|
||
}
|
||
|
||
d.leases[roomID] = next
|
||
|
||
return next, nil
|
||
}
|
||
|
||
// VerifyOwner 校验调用方手里的 lease token 仍是当前有效 owner。
|
||
func (d *MemoryDirectory) VerifyOwner(_ context.Context, roomID string, nodeID string, leaseToken string, now time.Time) (bool, error) {
|
||
d.mu.Lock()
|
||
defer d.mu.Unlock()
|
||
|
||
current, ok := d.leases[roomID]
|
||
if !ok {
|
||
return false, nil
|
||
}
|
||
|
||
return current.NodeID == nodeID && current.LeaseToken == leaseToken && current.ValidAt(now), nil
|
||
}
|
||
|
||
// ReleaseOwner 删除当前节点持有的内存 lease；其他节点持有时保持不变。
|
||
func (d *MemoryDirectory) ReleaseOwner(_ context.Context, roomID string, nodeID string) (bool, error) {
|
||
d.mu.Lock()
|
||
defer d.mu.Unlock()
|
||
|
||
current, ok := d.leases[roomID]
|
||
if !ok || current.NodeID != nodeID {
|
||
return false, nil
|
||
}
|
||
|
||
delete(d.leases, roomID)
|
||
return true, nil
|
||
}
|
||
|
||
// ForceExpire 只用于测试时模拟租约自然过期。
|
||
func (d *MemoryDirectory) ForceExpire(roomID string) {
|
||
d.mu.Lock()
|
||
defer d.mu.Unlock()
|
||
|
||
lease, ok := d.leases[roomID]
|
||
if !ok {
|
||
// 没有租约时无需处理，测试可安全重复调用。
|
||
return
|
||
}
|
||
|
||
lease.ExpiresAt = time.Unix(0, 0)
|
||
d.leases[roomID] = lease
|
||
}
|
||
|
||
// RegisterNode 写入测试用节点地址。
|
||
func (d *MemoryDirectory) RegisterNode(_ context.Context, node NodeRegistration, ttl time.Duration) error {
|
||
d.mu.Lock()
|
||
defer d.mu.Unlock()
|
||
|
||
now := time.Now().UTC()
|
||
if ttl <= 0 {
|
||
ttl = 30 * time.Second
|
||
}
|
||
node.UpdatedAt = now
|
||
node.ExpiresAt = now.Add(ttl)
|
||
d.nodes[node.NodeID] = node
|
||
return nil
|
||
}
|
||
|
||
// LookupNode 查询测试用节点地址，过期注册按不存在处理。
|
||
func (d *MemoryDirectory) LookupNode(_ context.Context, nodeID string) (NodeRegistration, bool, error) {
|
||
d.mu.Lock()
|
||
defer d.mu.Unlock()
|
||
|
||
node, ok := d.nodes[nodeID]
|
||
if !ok || !node.ValidAt(time.Now().UTC()) {
|
||
return NodeRegistration{}, false, nil
|
||
}
|
||
|
||
return node, true, nil
|
||
}
|
||
|
||
// UnregisterNode 删除测试用节点地址。
|
||
func (d *MemoryDirectory) UnregisterNode(_ context.Context, nodeID string) error {
|
||
d.mu.Lock()
|
||
defer d.mu.Unlock()
|
||
|
||
delete(d.nodes, nodeID)
|
||
return nil
|
||
}
|