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, 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() }