diff --git a/services/room-service/internal/room/service/admin_robot_room.go b/services/room-service/internal/room/service/admin_robot_room.go index 45455c2c..e234d87a 100644 --- a/services/room-service/internal/room/service/admin_robot_room.go +++ b/services/room-service/internal/room/service/admin_robot_room.go @@ -278,9 +278,37 @@ func (s *Service) startActiveRobotRooms(ctx context.Context) { return } logx.Info(ctx, "robot_room_runtime_scan_completed", slog.Int("active_room_count", len(configs))) + // 在启动/停止决策前先记录已运行集合;扫描期间由管理端并发新建的房间不在本轮 active 列表里, + // 不能被误判为已停用。 + runningRoomIDs := s.runningRobotRoomIDs() + activeRoomIDs := make(map[string]bool, len(configs)) for _, config := range configs { + activeRoomIDs[config.RoomID] = true + ownerNodeID, ownedElsewhere, err := s.roomLeaseOwnedElsewhere(ctx, config.AppCode, config.RoomID) + if err != nil { + // 目录抖动时保持现状:不认领也不停止,等下一轮扫描重新判定。 + logx.Warn(ctx, "robot_room_runtime_lease_lookup_failed", slog.String("room_id", config.RoomID), slog.String("error", err.Error())) + continue + } + if ownedElsewhere { + // 其他节点持有有效 lease 时,本节点对该房间的任何命令都只会得到 CONFLICT; + // 不启动新循环,并停掉执行权已迁走后仍残留的本地循环。 + if s.stopRobotRoomRuntime(config.RoomID) { + logx.Warn(ctx, "robot_room_runtime_ownership_lost", slog.String("room_id", config.RoomID), slog.String("owner_node_id", ownerNodeID)) + } + continue + } s.startRobotRoomRuntime(ctx, config) } + for _, roomID := range runningRoomIDs { + if activeRoomIDs[roomID] { + continue + } + // 停用/删除请求可能落在其他节点;配置不再 active 的房间不能在任何节点继续跑循环。 + if s.stopRobotRoomRuntime(roomID) { + logx.Info(ctx, "robot_room_runtime_stopped_inactive", slog.String("room_id", roomID)) + } + } } func (s *Service) startRobotRoomRuntime(parent context.Context, config RobotRoomConfig) { @@ -314,6 +342,23 @@ func (s *Service) startRobotRoomRuntime(parent context.Context, config RobotRoom } s.robotRuntimeMu.Unlock() + owned, ownerNodeID, err := s.claimRobotRoomOwnership(parent, config) + if err != nil { + logx.Warn(parent, "robot_room_runtime_lease_claim_failed", slog.String("room_id", config.RoomID), slog.String("error", err.Error())) + return + } + if !owned { + // 机器人房循环只能在 lease owner 节点运行;非 owner 节点启动只会对 owner 的房间刷 CONFLICT。 + logx.Info(parent, "robot_room_runtime_start_skipped", + slog.String("room_id", config.RoomID), + slog.String("status", config.Status), + slog.Int("robot_count", len(config.RobotUserIDs)), + slog.String("reason", "not_owner"), + slog.String("owner_node_id", ownerNodeID), + ) + return + } + if err := s.ensureRobotRoomParticipants(parent, config); err != nil { logx.Warn(parent, "robot_room_runtime_prepare_failed", slog.String("room_id", config.RoomID), slog.String("error", err.Error())) return @@ -339,16 +384,97 @@ func (s *Service) startRobotRoomRuntime(parent context.Context, config RobotRoom defer s.clearRobotRoomRuntime(config.RoomID, token) s.runRobotRoomRuntime(ctx, config) }() + go s.runRobotRoomLeaseKeeper(ctx, config, token) } -func (s *Service) stopRobotRoomRuntime(roomID string) { +// claimRobotRoomOwnership 在无有效 owner 或本节点已是 owner 时认领/续租机器人房 lease; +// 其他节点持有有效 lease 时返回现任 owner,调用方必须放弃启动循环。 +func (s *Service) claimRobotRoomOwnership(ctx context.Context, config RobotRoomConfig) (bool, string, error) { + if s.directory == nil { + // 测试中的极简 Service 可能没有装配 directory;生产 HealthCheck 会拒绝这种配置。 + return true, "", nil + } + lease, err := s.directory.EnsureOwner(ctx, runtimeRoomKey(config.AppCode, config.RoomID), s.nodeID, s.clock.Now(), s.leaseTTL) + if err != nil { + return false, "", err + } + if lease.NodeID != s.nodeID { + return false, lease.NodeID, nil + } + return true, "", nil +} + +// runRobotRoomLeaseKeeper 以短于 leaseTTL 的周期续租机器人房 lease。 +// 机器人房的写命令间隔可能远大于 leaseTTL,只靠命令续租会让 lease 频繁过期、被其他节点的 +// 扫描抢走;续租发现执行权已被接管时立即停掉本地 runtime,不再对新 owner 的房间空转 CONFLICT。 +func (s *Service) runRobotRoomLeaseKeeper(ctx context.Context, config RobotRoomConfig, token string) { + if s.directory == nil { + return + } + interval := s.leaseTTL / 3 + if interval <= 0 { + interval = time.Second + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + lease, err := s.directory.EnsureOwner(ctx, runtimeRoomKey(config.AppCode, config.RoomID), s.nodeID, s.clock.Now(), s.leaseTTL) + if err != nil { + // 目录抖动时保留 runtime 等下一次续租;期间的房间命令也会经 ensureCell 续租。 + logx.Warn(ctx, "robot_room_lease_renew_failed", slog.String("room_id", config.RoomID), slog.String("error", err.Error())) + continue + } + if lease.NodeID != s.nodeID { + logx.Warn(ctx, "robot_room_runtime_ownership_lost", + slog.String("room_id", config.RoomID), + slog.String("owner_node_id", lease.NodeID), + ) + s.stopRobotRoomRuntimeToken(config.RoomID, token) + return + } + } +} + +// stopRobotRoomRuntime 停止指定房间的本地循环;返回是否确实存在运行中的 runtime。 +func (s *Service) stopRobotRoomRuntime(roomID string) bool { s.robotRuntimeMu.Lock() - runtime := s.robotRuntimes[roomID] + runtime, exists := s.robotRuntimes[roomID] delete(s.robotRuntimes, roomID) s.robotRuntimeMu.Unlock() if runtime.cancel != nil { runtime.cancel() } + return exists +} + +// stopRobotRoomRuntimeToken 只停止仍由该 token 代表的 runtime,避免误停同房间的新一轮启动。 +func (s *Service) stopRobotRoomRuntimeToken(roomID string, token string) { + s.robotRuntimeMu.Lock() + runtime, exists := s.robotRuntimes[roomID] + if !exists || runtime.token != token { + s.robotRuntimeMu.Unlock() + return + } + delete(s.robotRuntimes, roomID) + s.robotRuntimeMu.Unlock() + if runtime.cancel != nil { + runtime.cancel() + } +} + +func (s *Service) runningRobotRoomIDs() []string { + s.robotRuntimeMu.Lock() + defer s.robotRuntimeMu.Unlock() + roomIDs := make([]string, 0, len(s.robotRuntimes)) + for roomID := range s.robotRuntimes { + roomIDs = append(roomIDs, roomID) + } + return roomIDs } func (s *Service) stopAllRobotRoomRuntimes() { diff --git a/services/room-service/internal/room/service/admin_robot_room_lease_test.go b/services/room-service/internal/room/service/admin_robot_room_lease_test.go new file mode 100644 index 00000000..77377610 --- /dev/null +++ b/services/room-service/internal/room/service/admin_robot_room_lease_test.go @@ -0,0 +1,184 @@ +package service + +import ( + "context" + "testing" + "time" + + "hyapp/pkg/appcode" + "hyapp/services/room-service/internal/router" +) + +type robotRoomScanRepository struct { + Repository + configs []RobotRoomConfig +} + +func (r *robotRoomScanRepository) ListActiveRobotRooms(context.Context) ([]RobotRoomConfig, error) { + return r.configs, nil +} + +func activeRobotRoomConfig(roomID string) RobotRoomConfig { + return RobotRoomConfig{ + AppCode: appcode.Default, + RoomID: roomID, + Status: robotRoomStatusActive, + RobotUserIDs: []int64{1001, 1002}, + } +} + +func TestClaimRobotRoomOwnershipRespectsOtherValidOwner(t *testing.T) { + ctx := context.Background() + now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC) + directory := router.NewMemoryDirectory() + svc := &Service{ + nodeID: "room-node-a", + leaseTTL: 10 * time.Second, + clock: fixedRoomTestClock{now: now}, + directory: directory, + } + config := activeRobotRoomConfig("robot-claim-test") + + owned, ownerNodeID, err := svc.claimRobotRoomOwnership(ctx, config) + if err != nil || !owned || ownerNodeID != "" { + t.Fatalf("unowned robot room should be claimed by this node: owned=%v owner=%q err=%v", owned, ownerNodeID, err) + } + if owned, _, err := svc.claimRobotRoomOwnership(ctx, config); err != nil || !owned { + t.Fatalf("self-owned robot room should renew: owned=%v err=%v", owned, err) + } + + directory.ForceExpire(runtimeRoomKey(config.AppCode, config.RoomID)) + if _, err := directory.EnsureOwner(ctx, runtimeRoomKey(config.AppCode, config.RoomID), "room-node-b", now, 10*time.Second); err != nil { + t.Fatalf("other node takeover: %v", err) + } + owned, ownerNodeID, err = svc.claimRobotRoomOwnership(ctx, config) + if err != nil || owned || ownerNodeID != "room-node-b" { + t.Fatalf("valid lease on another node must not be claimed: owned=%v owner=%q err=%v", owned, ownerNodeID, err) + } +} + +func TestStartRobotRoomRuntimeSkipsWhenOwnedElsewhere(t *testing.T) { + ctx := context.Background() + now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC) + directory := router.NewMemoryDirectory() + svc := &Service{ + nodeID: "room-node-a", + leaseTTL: 10 * time.Second, + clock: fixedRoomTestClock{now: now}, + directory: directory, + } + config := activeRobotRoomConfig("robot-owned-elsewhere") + if _, err := directory.EnsureOwner(ctx, runtimeRoomKey(config.AppCode, config.RoomID), "room-node-b", now, 10*time.Second); err != nil { + t.Fatalf("seed other-node lease: %v", err) + } + + // repository 故意留空:非 owner 分支必须在触达任何房间命令/仓储调用前返回。 + svc.startRobotRoomRuntime(ctx, config) + if len(svc.robotRuntimes) != 0 { + t.Fatalf("non-owner node must not start robot room runtime, got %d", len(svc.robotRuntimes)) + } +} + +func TestStartActiveRobotRoomsReconcilesOwnershipAndInactiveRooms(t *testing.T) { + ctx := context.Background() + now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC) + directory := router.NewMemoryDirectory() + ownedConfig := activeRobotRoomConfig("robot-owned") + lostConfig := activeRobotRoomConfig("robot-lease-lost") + + lostCancelled := false + staleCancelled := false + ownedCancelled := false + svc := &Service{ + nodeID: "room-node-a", + leaseTTL: 10 * time.Second, + clock: fixedRoomTestClock{now: now}, + directory: directory, + repository: &robotRoomScanRepository{configs: []RobotRoomConfig{ownedConfig, lostConfig}}, + robotRuntimes: map[string]robotRoomRuntime{ + ownedConfig.RoomID: {cancel: func() { ownedCancelled = true }, token: "owned-token"}, + lostConfig.RoomID: {cancel: func() { lostCancelled = true }, token: "lost-token"}, + "robot-stale": {cancel: func() { staleCancelled = true }, token: "stale-token"}, + }, + } + if _, err := directory.EnsureOwner(ctx, runtimeRoomKey(ownedConfig.AppCode, ownedConfig.RoomID), svc.nodeID, now, svc.leaseTTL); err != nil { + t.Fatalf("seed self lease: %v", err) + } + if _, err := directory.EnsureOwner(ctx, runtimeRoomKey(lostConfig.AppCode, lostConfig.RoomID), "room-node-b", now, svc.leaseTTL); err != nil { + t.Fatalf("seed other-node lease: %v", err) + } + + svc.startActiveRobotRooms(ctx) + + if _, exists := svc.robotRuntimes[ownedConfig.RoomID]; !exists || ownedCancelled { + t.Fatalf("runtime owned by this node must keep running: exists=%v cancelled=%v", exists, ownedCancelled) + } + if _, exists := svc.robotRuntimes[lostConfig.RoomID]; exists || !lostCancelled { + t.Fatalf("runtime whose lease moved to another node must stop: exists=%v cancelled=%v", exists, lostCancelled) + } + if _, exists := svc.robotRuntimes["robot-stale"]; exists || !staleCancelled { + t.Fatalf("runtime without active config must stop: exists=%v cancelled=%v", exists, staleCancelled) + } +} + +func TestRobotRoomLeaseKeeperStopsRuntimeWhenOwnershipMoves(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC) + directory := router.NewMemoryDirectory() + svc := &Service{ + nodeID: "room-node-a", + leaseTTL: 30 * time.Millisecond, + clock: fixedRoomTestClock{now: now}, + directory: directory, + } + config := activeRobotRoomConfig("robot-keeper") + key := runtimeRoomKey(config.AppCode, config.RoomID) + if _, err := directory.EnsureOwner(ctx, key, svc.nodeID, now, svc.leaseTTL); err != nil { + t.Fatalf("seed self lease: %v", err) + } + + runtimeCancelled := make(chan struct{}) + svc.robotRuntimes = map[string]robotRoomRuntime{ + config.RoomID: {cancel: func() { close(runtimeCancelled) }, token: "keeper-token"}, + } + go svc.runRobotRoomLeaseKeeper(ctx, config, "keeper-token") + + // 续租周期内 keeper 必须保持 owner 且不打断 runtime。 + time.Sleep(50 * time.Millisecond) + select { + case <-runtimeCancelled: + t.Fatal("keeper must not stop runtime while this node still owns the lease") + default: + } + if lease, exists, err := directory.Lookup(ctx, key); err != nil || !exists || lease.NodeID != svc.nodeID || !lease.ValidAt(now) { + t.Fatalf("keeper should keep renewing the lease: lease=%+v exists=%v err=%v", lease, exists, err) + } + + // keeper 会在 ForceExpire 和另一节点抢占之间继续续租,重试直到 node-b 真正拿到有效 lease。 + takeoverDeadline := time.Now().Add(2 * time.Second) + for { + directory.ForceExpire(key) + if _, err := directory.EnsureOwner(ctx, key, "room-node-b", now, time.Minute); err != nil { + t.Fatalf("other node takeover: %v", err) + } + if lease, exists, err := directory.Lookup(ctx, key); err == nil && exists && lease.NodeID == "room-node-b" { + break + } + if time.Now().After(takeoverDeadline) { + t.Fatal("other node could not take over the lease") + } + } + + select { + case <-runtimeCancelled: + case <-time.After(2 * time.Second): + t.Fatal("keeper must stop the local runtime after ownership moved to another node") + } + svc.robotRuntimeMu.Lock() + _, exists := svc.robotRuntimes[config.RoomID] + svc.robotRuntimeMu.Unlock() + if exists { + t.Fatal("keeper must remove the runtime entry after ownership loss") + } +} diff --git a/services/room-service/internal/room/service/create_room.go b/services/room-service/internal/room/service/create_room.go index 6b5334b1..e36a1b6f 100644 --- a/services/room-service/internal/room/service/create_room.go +++ b/services/room-service/internal/room/service/create_room.go @@ -145,7 +145,7 @@ func (s *Service) CreateRoom(ctx context.Context, req *roomv1.CreateRoomRequest) if lease.NodeID != s.nodeID { // 有其他有效 owner 时当前节点不能创建或接管该房间。 - return nil, xerr.New(xerr.Conflict, fmt.Sprintf("room is owned by %s", lease.NodeID)) + return nil, roomOwnedByOtherNodeError(lease.NodeID) } if _, exists, err := s.repository.GetRoomMeta(ctx, cmd.RoomID()); err != nil { diff --git a/services/room-service/internal/room/service/gift_operation.go b/services/room-service/internal/room/service/gift_operation.go index 6d74ca9b..2f32b009 100644 --- a/services/room-service/internal/room/service/gift_operation.go +++ b/services/room-service/internal/room/service/gift_operation.go @@ -203,8 +203,9 @@ func (f *giftFlow) handleGiftOperationFailure(err error) { ctx, cancel := context.WithTimeout(context.WithoutCancel(f.ctx), giftOperationPersistTimeout) defer cancel() now := f.s.clock.Now() - if !f.debitCompleted && !giftOperationErrorRetryable(err) { + if !f.debitCompleted && !giftOperationErrorRetryable(err) && !isRoomOwnedByOtherNodeError(err) { // wallet 明确拒绝(余额不足、配置非法等)保证没有扣款,pending 可以安全关闭;同载荷的用户主动重试会重新激活。 + // 跨节点执行权冲突不在此列:它只说明本节点不是 owner,首次尝试可能已经扣款,必须留给 owner 节点重试发现回执。 if persistErr := f.s.repository.MarkGiftOperationCompensated(ctx, f.cmd.RoomID(), f.cmd.ID(), err.Error(), now.UnixMilli()); persistErr != nil { slog.WarnContext(ctx, "mark gift operation compensated failed", "room_id", f.cmd.RoomID(), "command_id", f.cmd.ID(), "error", persistErr) } @@ -334,6 +335,15 @@ func (s *Service) recoverGiftOperation(workerCtx context.Context, operation Gift defer cancel() now := s.clock.Now() + if ownerNodeID, ownedElsewhere, err := s.roomLeaseOwnedElsewhere(operationCtx, operation.AppCode, operation.RoomID); err == nil && ownedElsewhere { + // 非 owner 节点执行恢复只会在 ensureCell 得到跨节点 CONFLICT;释放 claim 让 owner 节点的 + // worker 在下一轮完成结算,避免恢复任务在错误节点空转。目录查询失败时按本地可执行处理, + // ensureCell 会再做权威判定。 + if persistErr := s.repository.MarkGiftOperationRetry(operationCtx, operation.RoomID, operation.CommandID, roomOwnedByOtherNodeMsgPrefix+ownerNodeID, now.Add(giftOperationRetryDelay).UnixMilli(), now.UnixMilli(), false); persistErr != nil { + slog.WarnContext(operationCtx, "defer gift operation to owner failed", "room_id", operation.RoomID, "command_id", operation.CommandID, "error", persistErr) + } + return + } var req roomv1.SendGiftRequest if err := proto.Unmarshal(operation.RequestPayload, &req); err != nil { _ = s.repository.MarkGiftOperationRetry(operationCtx, operation.RoomID, operation.CommandID, "stored request decode failed", now.Add(time.Minute).UnixMilli(), now.UnixMilli(), false) diff --git a/services/room-service/internal/room/service/gift_operation_internal_test.go b/services/room-service/internal/room/service/gift_operation_internal_test.go index 1abdc42f..b509d65b 100644 --- a/services/room-service/internal/room/service/gift_operation_internal_test.go +++ b/services/room-service/internal/room/service/gift_operation_internal_test.go @@ -13,8 +13,32 @@ import ( "hyapp/pkg/appcode" "hyapp/pkg/xerr" "hyapp/services/room-service/internal/room/command" + "hyapp/services/room-service/internal/router" ) +type giftOperationOwnershipRepository struct { + Repository + retries []giftOperationRetryCall + compensated []giftOperationRetryCall +} + +type giftOperationRetryCall struct { + roomID string + commandID string + lastError string + preserveLease bool +} + +func (r *giftOperationOwnershipRepository) MarkGiftOperationRetry(_ context.Context, roomID string, commandID string, lastError string, _ int64, _ int64, preserveLease bool) error { + r.retries = append(r.retries, giftOperationRetryCall{roomID: roomID, commandID: commandID, lastError: lastError, preserveLease: preserveLease}) + return nil +} + +func (r *giftOperationOwnershipRepository) MarkGiftOperationCompensated(_ context.Context, roomID string, commandID string, lastError string, _ int64) error { + r.compensated = append(r.compensated, giftOperationRetryCall{roomID: roomID, commandID: commandID, lastError: lastError}) + return nil +} + type compensatedGiftOperationRepository struct { Repository operation GiftOperation @@ -64,6 +88,88 @@ func TestGiftOperationRecoveryDeadlineKeepsClaimUntilCellCommitBudgetEnds(t *tes } } +func TestGiftOperationFailureOwnershipConflictRetriesInsteadOfCompensating(t *testing.T) { + ctx := appcode.WithContext(context.Background(), appcode.Default) + now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC) + req := &roomv1.SendGiftRequest{ + Meta: &roomv1.RequestMeta{AppCode: appcode.Default, RoomId: "room-owner-conflict", CommandId: "cmd-owner-conflict", ActorUserId: 101}, + TargetType: "user", + TargetUserId: 202, + GiftId: "rose", + GiftCount: 1, + } + + repository := &giftOperationOwnershipRepository{} + flow := newGiftFlow(&Service{repository: repository, clock: fixedRoomTestClock{now: now}}, ctx, req, giftSendOptions{}) + flow.operationStarted = true + flow.recovering = true + flow.handleGiftOperationFailure(roomOwnedByOtherNodeError("room-node-b")) + if len(repository.compensated) != 0 { + t.Fatalf("cross-node ownership conflict must not compensate a pending saga: %+v", repository.compensated) + } + if len(repository.retries) != 1 || repository.retries[0].preserveLease { + t.Fatalf("ownership conflict should release the claim for the owner node to retry: %+v", repository.retries) + } + + repository = &giftOperationOwnershipRepository{} + flow = newGiftFlow(&Service{repository: repository, clock: fixedRoomTestClock{now: now}}, ctx, req, giftSendOptions{}) + flow.operationStarted = true + flow.handleGiftOperationFailure(xerr.New(xerr.Conflict, "command_id payload conflict")) + if len(repository.compensated) != 1 || len(repository.retries) != 0 { + t.Fatalf("business conflict before debit must stay compensated: compensated=%+v retries=%+v", repository.compensated, repository.retries) + } +} + +func TestRecoverGiftOperationDefersToOwnerNode(t *testing.T) { + ctx := context.Background() + now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC) + directory := router.NewMemoryDirectory() + repository := &giftOperationOwnershipRepository{} + svc := &Service{ + nodeID: "room-node-a", + leaseTTL: 10 * time.Second, + clock: fixedRoomTestClock{now: now}, + directory: directory, + repository: repository, + } + operation := GiftOperation{ + AppCode: appcode.Default, + RoomID: "room-recovery-owner", + CommandID: "cmd-recovery-owner", + Status: GiftOperationStatusPending, + } + if _, err := directory.EnsureOwner(ctx, runtimeRoomKey(operation.AppCode, operation.RoomID), "room-node-b", now, 10*time.Second); err != nil { + t.Fatalf("seed other-node lease: %v", err) + } + + // RequestPayload 故意留空:非 owner 分支必须在解码存量请求、进入 Room Cell 之前把任务让给 owner。 + svc.recoverGiftOperation(ctx, operation) + if len(repository.retries) != 1 { + t.Fatalf("non-owner recovery must defer exactly once, got %+v", repository.retries) + } + if repository.retries[0].preserveLease || repository.retries[0].lastError != roomOwnedByOtherNodeMsgPrefix+"room-node-b" { + t.Fatalf("defer must release the claim and record the current owner: %+v", repository.retries[0]) + } + if len(repository.compensated) != 0 { + t.Fatalf("defer must not compensate the operation: %+v", repository.compensated) + } +} + +func TestIsRoomOwnedByOtherNodeError(t *testing.T) { + if !isRoomOwnedByOtherNodeError(roomOwnedByOtherNodeError("room-node-b")) { + t.Fatal("ownership conflict error must be recognized") + } + for _, err := range []error{ + xerr.New(xerr.Conflict, "command_id payload conflict"), + xerr.New(xerr.Unavailable, roomOwnedByOtherNodeMsgPrefix+"room-node-b"), + errors.New(roomOwnedByOtherNodeMsgPrefix + "room-node-b"), + } { + if isRoomOwnedByOtherNodeError(err) { + t.Fatalf("must not treat %v as cross-node ownership conflict", err) + } + } +} + func TestCompensatedGiftOperationReactivatesOnlyAfterPayloadMatch(t *testing.T) { ctx := appcode.WithContext(context.Background(), appcode.Default) originalRequest := &roomv1.SendGiftRequest{ diff --git a/services/room-service/internal/room/service/helpers.go b/services/room-service/internal/room/service/helpers.go index 67e31e0e..3d4a6e33 100644 --- a/services/room-service/internal/room/service/helpers.go +++ b/services/room-service/internal/room/service/helpers.go @@ -130,6 +130,22 @@ func runtimeRoomKeyFromContext(ctx context.Context, roomID string) string { return runtimeRoomKey(appcode.FromContext(ctx), roomID) } +// roomLeaseOwnedElsewhere 判断房间执行权当前是否由其他节点持有有效 lease;只读判定,不接管也不续租。 +func (s *Service) roomLeaseOwnedElsewhere(ctx context.Context, appCode string, roomID string) (string, bool, error) { + if s.directory == nil { + // 测试中的极简 Service 可能没有装配 directory;生产 HealthCheck 会拒绝这种配置。 + return "", false, nil + } + lease, exists, err := s.directory.Lookup(ctx, runtimeRoomKey(appCode, roomID)) + if err != nil { + return "", false, err + } + if exists && lease.ValidAt(s.clock.Now()) && lease.NodeID != s.nodeID { + return lease.NodeID, true, nil + } + return "", false, nil +} + func (s *Service) loadedRoomLease(ctx context.Context, roomRef loadedRoomRef, now time.Time) (router.Lease, bool, error) { // 后台 worker 通过 EnsureOwner 续租当前 Cell,不能在没有 lease 的情况下读取旧内存态。 lease, err := s.directory.EnsureOwner(ctx, runtimeRoomKey(roomRef.AppCode, roomRef.RoomID), s.nodeID, now, s.leaseTTL) diff --git a/services/room-service/internal/room/service/recovery.go b/services/room-service/internal/room/service/recovery.go index b39dbddc..7845f32d 100644 --- a/services/room-service/internal/room/service/recovery.go +++ b/services/room-service/internal/room/service/recovery.go @@ -3,6 +3,7 @@ package service import ( "context" "fmt" + "strings" "google.golang.org/protobuf/proto" roomv1 "hyapp.local/api/proto/room/v1" @@ -14,6 +15,21 @@ import ( "hyapp/services/room-service/internal/router" ) +// roomOwnedByOtherNodeMsgPrefix 是跨节点执行权冲突的稳定错误文案前缀; +// isRoomOwnedByOtherNodeError 依赖它把这类短暂冲突和业务终态 Conflict 区分开。 +const roomOwnedByOtherNodeMsgPrefix = "room is owned by " + +// roomOwnedByOtherNodeError 表示房间执行权由其他节点的有效 lease 持有。 +func roomOwnedByOtherNodeError(nodeID string) error { + return xerr.New(xerr.Conflict, roomOwnedByOtherNodeMsgPrefix+nodeID) +} + +// isRoomOwnedByOtherNodeError 判断错误是否为跨节点执行权冲突。 +// 它只在 lease TTL 窗口内成立,不代表业务不可重试;补偿判定不能把它当作终态。 +func isRoomOwnedByOtherNodeError(err error) bool { + return xerr.CodeOf(err) == xerr.Conflict && strings.HasPrefix(xerr.MessageOf(err), roomOwnedByOtherNodeMsgPrefix) +} + // ensureCell 确保当前节点拥有房间执行权,并在本地缺少 Cell 时从 MySQL 恢复。 func (s *Service) ensureCell(ctx context.Context, roomID string) (*cell.RoomCell, router.Lease, error) { // 每次写命令先刷新或接管 lease,确保同一房间同一时刻只有一个执行节点。 @@ -24,7 +40,7 @@ func (s *Service) ensureCell(ctx context.Context, roomID string) (*cell.RoomCell if lease.NodeID != s.nodeID { // 仍有其他有效 owner 时当前节点不能执行该房间命令。 - return nil, router.Lease{}, xerr.New(xerr.Conflict, fmt.Sprintf("room is owned by %s", lease.NodeID)) + return nil, router.Lease{}, roomOwnedByOtherNodeError(lease.NodeID) } if roomCell := s.loadCellForLease(ctx, roomID, lease); roomCell != nil {