320 lines
13 KiB
Go
320 lines
13 KiB
Go
package service
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"time"
|
||
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
"hyapp/services/room-service/internal/room/command"
|
||
"hyapp/services/room-service/internal/room/outbox"
|
||
"hyapp/services/room-service/internal/room/rank"
|
||
"hyapp/services/room-service/internal/room/state"
|
||
"hyapp/services/room-service/internal/router"
|
||
)
|
||
|
||
const roomMutationCommitTimeout = 5 * time.Second
|
||
|
||
func (s *Service) mutateRoom(ctx context.Context, cmd command.Command, replayable bool, mutate func(now time.Time, current *state.RoomState, localRank *rank.LocalRank) (mutationResult, []outbox.Record, error)) (mutationResult, error) {
|
||
startedAt := time.Now()
|
||
var idempotencyMS int64
|
||
var leaseMS int64
|
||
var saveMutationMS int64
|
||
var snapshotMS int64
|
||
var projectionMS int64
|
||
if cmd.RoomID() == "" || cmd.ID() == "" || cmd.ActorUserID() == 0 {
|
||
// command meta 是幂等、路由、审计和权限判断的共同基础，缺一不可。
|
||
return mutationResult{}, xerr.New(xerr.InvalidArgument, "command meta is incomplete")
|
||
}
|
||
if s.isDraining() {
|
||
// 下线中的节点只让已经进入执行链路的命令收尾，不再接收新的 Room Cell 命令。
|
||
return mutationResult{}, xerr.New(xerr.Unavailable, "room service is draining")
|
||
}
|
||
|
||
idempotencyStartedAt := time.Now()
|
||
payload, seen, err := s.commandPayloadForIdempotency(ctx, cmd)
|
||
idempotencyMS = elapsedMS(idempotencyStartedAt)
|
||
if err != nil {
|
||
return mutationResult{}, err
|
||
}
|
||
if seen {
|
||
// 幂等命令不重新执行 mutate，也不会再次调用 wallet 或写 outbox。
|
||
snapshot, err := s.currentSnapshot(ctx, cmd.RoomID())
|
||
if err != nil {
|
||
return mutationResult{}, err
|
||
}
|
||
|
||
return mutationResult{snapshot: snapshot}, nil
|
||
}
|
||
|
||
// 确保当前节点持有房间 lease，并在没有内存 Cell 时完成恢复。
|
||
leaseStartedAt := time.Now()
|
||
roomCell, lease, err := s.ensureCell(ctx, cmd.RoomID())
|
||
leaseMS = elapsedMS(leaseStartedAt)
|
||
if err != nil {
|
||
return mutationResult{}, err
|
||
}
|
||
|
||
value, err := roomCell.Do(ctx, func(current *state.RoomState, currentRank *rank.LocalRank) (any, error) {
|
||
// 在副本上执行命令，只有 command log/outbox 保存成功后才整体替换当前状态。
|
||
nextState := current.Clone()
|
||
nextRank := currentRank.Clone()
|
||
now := s.clock.Now()
|
||
|
||
result, outboxRecords, err := mutate(now, nextState, nextRank)
|
||
if err != nil {
|
||
// 校验、钱包或业务失败时不提交状态，不写 command log。
|
||
return nil, err
|
||
}
|
||
|
||
if result.snapshot == nil {
|
||
// 无特殊返回需求的命令默认返回变更后的快照。
|
||
result.snapshot = nextState.ToProto()
|
||
}
|
||
result.snapshot = snapshotWithApp(ctx, result.snapshot)
|
||
|
||
if result.snapshot.GetVersion() != current.Version {
|
||
// 版本变化代表本命令产生实际状态变更，需要持久化命令和事件。
|
||
if err := s.ensureLeaseStillOwned(ctx, cmd.RoomID(), lease); err != nil {
|
||
return nil, err
|
||
}
|
||
// 机器人展示只是 App 视觉补偿，提交前按短窗口降采样；命令日志、房间状态和主 outbox 不受影响。
|
||
robotDisplayRecords := s.sampleRobotDisplayRecords(result.robotDisplayRecords)
|
||
durableOutboxRecords, directIMRecords := splitRoomDirectIMRecords(outboxRecords, result.suppressDirectIMEventTypes)
|
||
// mutate 可额外返回只服务客户端展示的直发 IM，典型场景是重复 JoinRoom 的入房飘窗；
|
||
// 这类事件不能进入 durable outbox，否则会污染活跃统计和补偿消费。
|
||
directIMRecords = append(directIMRecords, result.directIMRecords...)
|
||
saveStartedAt := time.Now()
|
||
commitCtx, commitCancel := roomMutationCommitContext(ctx)
|
||
err := s.repository.SaveMutation(commitCtx, MutationCommit{
|
||
Command: CommandRecord{
|
||
AppCode: appcode.FromContext(ctx),
|
||
RoomID: cmd.RoomID(),
|
||
RoomVersion: nextState.Version,
|
||
CommandID: cmd.ID(),
|
||
CommandType: cmd.Type(),
|
||
Payload: commandPayloadForRecord(payload, result),
|
||
Replayable: replayable,
|
||
OwnerNodeID: s.nodeID,
|
||
LeaseToken: lease.LeaseToken,
|
||
CreatedAtMS: now.UnixMilli(),
|
||
},
|
||
OutboxRecords: durableOutboxRecords,
|
||
DeleteRoom: result.deleteRoom,
|
||
RoomStatus: result.roomStatus,
|
||
RoomSeatCount: result.roomSeatCount,
|
||
RoomPasswordHash: result.roomPasswordHash,
|
||
RoomMode: result.roomMode,
|
||
VisibleRegionID: result.visibleRegionID,
|
||
CloseInfo: result.closeInfo,
|
||
RoomUserGiftStats: result.roomUserGiftStats,
|
||
RoomUserContributionPeriodStats: result.roomUserContributionPeriodStats,
|
||
RoomWeeklyContribution: result.roomWeeklyContribution,
|
||
})
|
||
commitCancel()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
saveMutationMS += elapsedMS(saveStartedAt)
|
||
// 保存已提交 outbox 记录，出 Cell 锁后由 outbox worker 异步投递 activity/MQ 业务事实，避免外部慢调用占住房间串行执行队列。
|
||
result.outboxRecords = durableOutboxRecords
|
||
result.robotDisplayRecords = robotDisplayRecords
|
||
result.directIMRecords = directIMRecords
|
||
// MySQL 事务成功提交后再发本进程唤醒信号；信号丢失只影响延迟，不影响后续周期扫描补偿。
|
||
s.notifyOutboxCommitted(durableOutboxRecords)
|
||
// 直接 IM lane 只承载展示事件；RoomRocketIgnited 同时保留 durable outbox，让延迟发射调度继续可重试。
|
||
s.publishDirectIMRecordsBestEffort(ctx, directIMRecords)
|
||
// 机器人展示事件允许极小概率丢失，不写 MySQL outbox；提交成功后后台直发 IM，失败只记录日志。
|
||
s.publishRobotDisplayRecordsBestEffort(ctx, robotDisplayRecords)
|
||
|
||
// 持久化成功后再替换内存状态，避免内存提交领先于恢复来源。
|
||
current.ReplaceWith(nextState)
|
||
currentRank.ReplaceWith(nextRank)
|
||
result.snapshot = snapshotWithApp(ctx, current.ToProto())
|
||
result.applied = true
|
||
} else {
|
||
// no-op 命令仍写入 command log，保证相同 command_id 重试可命中幂等，
|
||
// 不同 payload 复用 command_id 会被后续入口拒绝为 CONFLICT。
|
||
if err := s.ensureLeaseStillOwned(ctx, cmd.RoomID(), lease); err != nil {
|
||
return nil, err
|
||
}
|
||
saveStartedAt := time.Now()
|
||
commitCtx, commitCancel := roomMutationCommitContext(ctx)
|
||
err := s.repository.SaveMutation(commitCtx, MutationCommit{
|
||
Command: CommandRecord{
|
||
AppCode: appcode.FromContext(ctx),
|
||
RoomID: cmd.RoomID(),
|
||
RoomVersion: current.Version,
|
||
CommandID: cmd.ID(),
|
||
CommandType: cmd.Type(),
|
||
Payload: payload,
|
||
Replayable: false,
|
||
OwnerNodeID: s.nodeID,
|
||
LeaseToken: lease.LeaseToken,
|
||
CreatedAtMS: now.UnixMilli(),
|
||
},
|
||
})
|
||
commitCancel()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
saveMutationMS += elapsedMS(saveStartedAt)
|
||
}
|
||
|
||
return result, nil
|
||
})
|
||
if err != nil {
|
||
return mutationResult{}, err
|
||
}
|
||
|
||
result := value.(mutationResult)
|
||
if result.applied && result.roomRocketProgress != nil && s.roomRocketProgressCoalescer != nil {
|
||
// 火箭进度合并必须发生在命令日志、快照来源和非火箭 outbox 已提交之后；未提交命令不能泄露展示事件。
|
||
s.roomRocketProgressCoalescer.enqueue(*result.roomRocketProgress)
|
||
}
|
||
|
||
if !result.deleteRoom && (result.applied || result.forceSnapshot) && result.snapshot != nil && (result.forceSnapshot || shouldSnapshot(result.snapshot, s.snapshotEveryN)) {
|
||
// 快照失败时返回错误，因为没有快照会增加恢复成本并可能影响 guard 查询。
|
||
snapshotStartedAt := time.Now()
|
||
if err := s.persistSnapshot(ctx, result.snapshot); err != nil {
|
||
return mutationResult{}, err
|
||
}
|
||
snapshotMS = elapsedMS(snapshotStartedAt)
|
||
}
|
||
if result.applied && !result.deleteRoom {
|
||
projectionStartedAt := time.Now()
|
||
// 列表读模型最终一致，失败只记录日志，不回滚已经提交的 Room Cell 命令。
|
||
s.projectRoomListBestEffort(ctx, result.snapshot)
|
||
// 当前房间恢复读模型同样最终一致；权威校验仍以 Room Cell 快照为准。
|
||
s.projectRoomPresenceForCommandBestEffort(ctx, cmd, result.snapshot, result.user)
|
||
// 跨房间贡献榜是 Redis 轻量读模型，不能影响已提交的房间状态和账务事实。
|
||
s.recordRoomGiftLeaderboardBestEffort(ctx, result.roomGiftLeaderboard)
|
||
projectionMS = elapsedMS(projectionStartedAt)
|
||
}
|
||
|
||
logRoomCommandTiming(ctx, cmd.RoomID(), cmd.ID(), cmd.Type(), result.applied, elapsedMS(startedAt), idempotencyMS, leaseMS, saveMutationMS, snapshotMS, projectionMS, result.walletDebitMS, 0, 0, len(result.outboxRecords) > 0, result.syncEvent != nil)
|
||
|
||
return result, nil
|
||
}
|
||
|
||
func roomMutationCommitContext(ctx context.Context) (context.Context, context.CancelFunc) {
|
||
// 钱包扣费等外部副作用完成后，command log/outbox 是恢复和补偿的权威来源；
|
||
// 这里保留 app_code 等 context values，但不继承 HTTP 断连取消信号，避免已扣费命令在提交阶段被调用方取消打断。
|
||
return context.WithTimeout(context.WithoutCancel(ctx), roomMutationCommitTimeout)
|
||
}
|
||
|
||
func (s *Service) commandPayloadForIdempotency(ctx context.Context, cmd command.Command) ([]byte, bool, error) {
|
||
// 先序列化当前命令，后续无论首次提交还是冲突校验都使用同一份 payload。
|
||
payload, err := command.Serialize(cmd)
|
||
if err != nil {
|
||
return nil, false, err
|
||
}
|
||
|
||
// command_id 在同一 room_id 内唯一；已有记录表示客户端重试或 command_id 被错误复用。
|
||
record, exists, err := s.repository.GetCommand(ctx, cmd.RoomID(), cmd.ID())
|
||
if err != nil {
|
||
return nil, false, err
|
||
}
|
||
if !exists {
|
||
return payload, false, nil
|
||
}
|
||
if cmd.Type() == (command.SetRoomPassword{}).Type() && record.CommandType == (command.SetRoomPassword{}).Type() {
|
||
// 密码命令的持久 payload 只有哈希，必须走专门逻辑比较 transient 明文和已存哈希。
|
||
matched, err := setRoomPasswordIdempotencyMatches(cmd, payload, record.Payload)
|
||
if err != nil {
|
||
return nil, true, err
|
||
}
|
||
if !matched {
|
||
return nil, true, xerr.New(xerr.Conflict, "command_id payload conflict")
|
||
}
|
||
return payload, true, nil
|
||
}
|
||
currentIDPayload, err := command.IdempotencyPayload(payload)
|
||
if err != nil {
|
||
return nil, true, err
|
||
}
|
||
storedIDPayload, err := command.IdempotencyPayload(record.Payload)
|
||
if err != nil {
|
||
return nil, true, err
|
||
}
|
||
if record.CommandType != cmd.Type() || !bytes.Equal(storedIDPayload, currentIDPayload) {
|
||
// 同一个 command_id 携带不同业务载荷必须显式冲突，不能返回旧命令结果掩盖客户端错误。
|
||
return nil, true, xerr.New(xerr.Conflict, "command_id payload conflict")
|
||
}
|
||
|
||
return payload, true, nil
|
||
}
|
||
|
||
func setRoomPasswordIdempotencyMatches(cmd command.Command, currentPayload []byte, storedPayload []byte) (bool, error) {
|
||
// 先比较去除哈希后的业务载荷，locked/room/actor 等字段不同仍然是冲突。
|
||
currentIDPayload, err := command.IdempotencyPayload(currentPayload)
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
storedIDPayload, err := command.IdempotencyPayload(storedPayload)
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
if !bytes.Equal(storedIDPayload, currentIDPayload) {
|
||
return false, nil
|
||
}
|
||
|
||
current, ok := setRoomPasswordCommand(cmd)
|
||
if !ok {
|
||
return false, nil
|
||
}
|
||
if !current.Locked {
|
||
// 解锁命令没有密码语义；通用 payload 相同即可视作同一动作重试。
|
||
return true, nil
|
||
}
|
||
stored, err := command.Deserialize((command.SetRoomPassword{}).Type(), storedPayload)
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
storedPassword, ok := stored.(*command.SetRoomPassword)
|
||
if !ok {
|
||
return false, nil
|
||
}
|
||
// bcrypt 哈希带随机盐，不能直接比较两次请求生成的哈希；只在当前请求内用 transient 明文校验已持久化哈希。
|
||
return roomPasswordMatches(storedPassword.PasswordHash, current.Password), nil
|
||
}
|
||
|
||
func setRoomPasswordCommand(cmd command.Command) (command.SetRoomPassword, bool) {
|
||
// mutateRoom 入口有时传值、有时传指针，幂等判断统一收敛成值对象。
|
||
switch typed := cmd.(type) {
|
||
case command.SetRoomPassword:
|
||
return typed, true
|
||
case *command.SetRoomPassword:
|
||
if typed == nil {
|
||
return command.SetRoomPassword{}, false
|
||
}
|
||
return *typed, true
|
||
default:
|
||
return command.SetRoomPassword{}, false
|
||
}
|
||
}
|
||
|
||
func commandPayloadForRecord(defaultPayload []byte, result mutationResult) []byte {
|
||
if len(result.commandPayload) > 0 {
|
||
// SendGift 等命令需要把外部结算后的确定性结果写进 command log，恢复时不能重算。
|
||
return result.commandPayload
|
||
}
|
||
return defaultPayload
|
||
}
|
||
|
||
func (s *Service) ensureLeaseStillOwned(ctx context.Context, roomID string, lease router.Lease) error {
|
||
ok, err := s.directory.VerifyOwner(ctx, runtimeRoomKeyFromContext(ctx, roomID), s.nodeID, lease.LeaseToken, s.clock.Now())
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !ok {
|
||
return xerr.New(xerr.Unavailable, "room lease is no longer owned by this node")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func elapsedMS(start time.Time) int64 {
|
||
return time.Since(start).Milliseconds()
|
||
}
|