diff --git a/services/room-service/internal/room/service/admin_robot_room_runtime_test.go b/services/room-service/internal/room/service/admin_robot_room_runtime_test.go index 39c0a0fc..9c1a710e 100644 --- a/services/room-service/internal/room/service/admin_robot_room_runtime_test.go +++ b/services/room-service/internal/room/service/admin_robot_room_runtime_test.go @@ -54,7 +54,10 @@ func TestRobotRoomStalePresenceKeepsConfiguredRobots(t *testing.T) { } } -func TestReplayRoomHeartbeatAfterLeaveIsIgnored(t *testing.T) { +func TestReplayRoomHeartbeatWithoutPresenceRecreatesIt(t *testing.T) { + // 线上心跳会对丢失的 presence 静默恢复后才提交命令日志；回放遇到"用户不存在"的 + // 已提交心跳必须重现恢复（含 Version++），并且绝不能让整个 Room Cell 恢复失败。 + // 旧语义的"忽略跳过"会让恢复出的 OnlineUsers/版本与快照谱系分叉。 current := &state.RoomState{ OnlineUsers: map[int64]*state.RoomUserState{}, } @@ -67,13 +70,14 @@ func TestReplayRoomHeartbeatAfterLeaveIsIgnored(t *testing.T) { }, }, 1781675913212) if err != nil { - t.Fatalf("stale heartbeat after leave must not break recovery: %v", err) + t.Fatalf("heartbeat without presence must not break recovery: %v", err) } - if len(current.OnlineUsers) != 0 { - t.Fatalf("stale heartbeat must not recreate presence: %+v", current.OnlineUsers) + recovered, exists := current.OnlineUsers[123456] + if !exists || recovered.Role != "audience" || recovered.LastSeenAtMS != 1781675912713 { + t.Fatalf("replayed heartbeat heal must recreate presence from SentAtMS: %+v", current.OnlineUsers) } - if current.Version != 0 { - t.Fatalf("ignored heartbeat must not advance room version, got %d", current.Version) + if current.Version != 1 { + t.Fatalf("replayed heartbeat heal must advance room version exactly once, got %d", current.Version) } } diff --git a/services/room-service/internal/room/service/presence.go b/services/room-service/internal/room/service/presence.go index 449c6c4e..b9b06c0f 100644 --- a/services/room-service/internal/room/service/presence.go +++ b/services/room-service/internal/room/service/presence.go @@ -360,8 +360,31 @@ func (s *Service) RoomHeartbeat(ctx context.Context, req *roomv1.RoomHeartbeatRe existing, exists := current.OnlineUsers[cmd.ActorUserID()] if !exists { - // 心跳只刷新已存在 presence，不能把已离房用户重新加入房间。 - return mutationResult{}, nil, xerr.New(xerr.NotFound, "user not in room") + // presence 已丢失但客户端仍在按房间节奏心跳，说明发生过 stale 清理或 owner + // 迁移这类服务端单方面清理；不恢复的话该用户会陷入 MicUp 404、RTC token 403 + // 的粘性失败，只能退房重进。这里按"心跳即在房"原地恢复。 + // 安全边界与 JoinRoom 对齐：room active 已在上方校验，活跃 ban（踢人/封禁） + // 已在上方拦截；锁房进入需要密码而心跳不携带密码，非房主不得借心跳绕过锁房， + // 维持原 NotFound 契约让客户端走完整 JoinRoom 补交密码。 + if current.RoomPasswordHash != "" && current.OwnerUserID != cmd.ActorUserID() { + return mutationResult{}, nil, xerr.New(xerr.NotFound, "user not in room") + } + // 与 JoinRoom presence_recovery 相同的静默语义：只补 Room Cell 状态，不发 + // sync IM/direct IM/durable outbox，避免重复入房横幅与统计重复计数。 + // LeaveRoom 后仍在途的迟到心跳会短暂恢复 presence，由 stale worker 在下一个 + // 清理窗口自然移除，期间同样不产生任何对外事件。 + current.OnlineUsers[cmd.ActorUserID()] = &state.RoomUserState{ + UserID: cmd.ActorUserID(), + Role: recoveredPresenceRole(current, cmd.ActorUserID()), + JoinedAtMS: now.UnixMilli(), + LastSeenAtMS: now.UnixMilli(), + } + current.Version++ + snapshot := current.ToProto() + return mutationResult{ + snapshot: snapshot, + user: findProtoUser(snapshot, cmd.ActorUserID()), + }, nil, nil } existing.LastSeenAtMS = now.UnixMilli() @@ -384,6 +407,20 @@ func (s *Service) RoomHeartbeat(ctx context.Context, req *roomv1.RoomHeartbeatRe }, nil } +// recoveredPresenceRole 推导心跳恢复 presence 时的展示角色。房间权限判定始终以 +// OwnerUserID/AdminUsers 权威字段为准（见 actorRole），这里只保证在线列表的房主/ +// 管理员徽标不因恢复而丢失；无法从权威字段证明的身份一律回落 audience，不存在提权。 +func recoveredPresenceRole(current *state.RoomState, userID int64) string { + switch { + case current.OwnerUserID == userID: + return "owner" + case current.AdminUsers[userID]: + return "admin" + default: + return "audience" + } +} + // LeaveRoom 把用户从房间业务态移出。 func (s *Service) LeaveRoom(ctx context.Context, req *roomv1.LeaveRoomRequest) (*roomv1.LeaveRoomResponse, error) { ctx = contextFromMeta(ctx, req.GetMeta()) diff --git a/services/room-service/internal/room/service/presence_heartbeat_recovery_test.go b/services/room-service/internal/room/service/presence_heartbeat_recovery_test.go new file mode 100644 index 00000000..bee84e04 --- /dev/null +++ b/services/room-service/internal/room/service/presence_heartbeat_recovery_test.go @@ -0,0 +1,227 @@ +package service_test + +import ( + "context" + "testing" + "time" + + roomv1 "hyapp.local/api/proto/room/v1" + "hyapp/pkg/appcode" + "hyapp/pkg/xerr" + "hyapp/services/room-service/internal/integration" + roomservice "hyapp/services/room-service/internal/room/service" + "hyapp/services/room-service/internal/router" + "hyapp/services/room-service/internal/testutil/mysqltest" +) + +func newHeartbeatRecoveryService(t *testing.T, repository *mysqltest.Repository, nodeID string, clk *fixedRoomRocketClock) *roomservice.Service { + t.Helper() + cfg := roomservice.Config{ + NodeID: nodeID, + LeaseTTL: 10 * time.Second, + RankLimit: 20, + SnapshotEveryN: 1, + } + if clk != nil { + cfg.Clock = clk + } + return roomservice.New(cfg, router.NewMemoryDirectory(), repository, followTestWallet{}, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher()) +} + +func createHeartbeatRecoveryRoom(t *testing.T, svc *roomservice.Service, roomID string, ownerID int64, shortID string) { + t.Helper() + if _, err := svc.CreateRoom(context.Background(), &roomv1.CreateRoomRequest{ + Meta: roomservice.NewRequestMeta(roomID, ownerID), + SeatCount: 10, + Mode: "voice", + VisibleRegionId: 8101, + RoomName: "Heartbeat Recovery", + RoomAvatar: testRoomCoverURL, + RoomShortId: shortID, + }); err != nil { + t.Fatalf("create heartbeat recovery room fixture failed: %v", err) + } +} + +func TestRoomHeartbeatRecoversClearedPresence(t *testing.T) { + ctx := context.Background() + repository := mysqltest.NewRepository(t) + svc := newHeartbeatRecoveryService(t, repository, "node-hb-recover-test", nil) + + roomID := "room-hb-recover" + ownerID := int64(7501) + viewerID := int64(7601) + createHeartbeatRecoveryRoom(t, svc, roomID, ownerID, "hb-recover") + if _, err := svc.JoinRoom(ctx, &roomv1.JoinRoomRequest{ + Meta: roomservice.NewRequestMeta(roomID, viewerID), + Role: "audience", + }); err != nil { + t.Fatalf("join viewer failed: %v", err) + } + // LeaveRoom 与 stale sweep 走同一条 presence 删除路径，等价模拟"服务端单方面清掉 presence"。 + if _, err := svc.LeaveRoom(ctx, &roomv1.LeaveRoomRequest{ + Meta: roomservice.NewRequestMeta(roomID, viewerID), + }); err != nil { + t.Fatalf("leave viewer failed: %v", err) + } + + outboxBefore := repository.CountRowsByRoom("room_outbox", roomID) + heartbeatResp, err := svc.RoomHeartbeat(ctx, &roomv1.RoomHeartbeatRequest{ + Meta: roomservice.NewRequestMeta(roomID, viewerID), + }) + if err != nil { + t.Fatalf("heartbeat must recover cleared presence: %v", err) + } + recovered := findRoomUser(heartbeatResp.GetRoom(), viewerID) + if recovered == nil || recovered.GetRole() != "audience" { + t.Fatalf("recovered presence must be back online as audience: %+v", heartbeatResp.GetRoom()) + } + if outboxAfter := repository.CountRowsByRoom("room_outbox", roomID); outboxAfter != outboxBefore { + // 恢复必须静默：不发 RoomUserJoined 等 durable outbox 事件，避免重复入房横幅与统计重复计数。 + t.Fatalf("silent recovery must not emit outbox events: before=%d after=%d", outboxBefore, outboxAfter) + } + + if _, err := svc.MicUp(ctx, &roomv1.MicUpRequest{ + Meta: roomservice.NewRequestMeta(roomID, viewerID), + SeatNo: 1, + }); err != nil { + t.Fatalf("mic up must succeed right after heartbeat recovery: %v", err) + } +} + +func TestRoomHeartbeatKeepsRejectingKickedUser(t *testing.T) { + ctx := context.Background() + repository := mysqltest.NewRepository(t) + svc := newHeartbeatRecoveryService(t, repository, "node-hb-kick-guard-test", nil) + + roomID := "room-hb-kick-guard" + ownerID := int64(7701) + viewerID := int64(7801) + createHeartbeatRecoveryRoom(t, svc, roomID, ownerID, "hb-kick-guard") + if _, err := svc.JoinRoom(ctx, &roomv1.JoinRoomRequest{ + Meta: roomservice.NewRequestMeta(roomID, viewerID), + Role: "audience", + }); err != nil { + t.Fatalf("join viewer failed: %v", err) + } + if _, err := svc.KickUser(ctx, &roomv1.KickUserRequest{ + Meta: roomservice.NewRequestMeta(roomID, ownerID), + TargetUserId: viewerID, + }); err != nil { + t.Fatalf("kick viewer failed: %v", err) + } + + _, err := svc.RoomHeartbeat(ctx, &roomv1.RoomHeartbeatRequest{ + Meta: roomservice.NewRequestMeta(roomID, viewerID), + }) + if err == nil || xerr.CodeOf(err) != xerr.PermissionDenied { + t.Fatalf("kicked user heartbeat must stay PermissionDenied, got: %v", err) + } + + presenceResp, err := svc.VerifyRoomPresence(ctx, &roomv1.VerifyRoomPresenceRequest{ + AppCode: appcode.Default, + RoomId: roomID, + UserId: viewerID, + }) + if err != nil { + t.Fatalf("verify presence after rejected heartbeat failed: %v", err) + } + if presenceResp.GetPresent() || presenceResp.GetReason() != "user_banned" { + t.Fatalf("kicked user must not regain presence through heartbeat: %+v", presenceResp) + } +} + +func TestRoomHeartbeatRecoveryAfterKickExpires(t *testing.T) { + ctx := context.Background() + repository := mysqltest.NewRepository(t) + clk := &fixedRoomRocketClock{now: time.Date(2026, 7, 13, 8, 0, 0, 0, time.UTC)} + svc := newHeartbeatRecoveryService(t, repository, "node-hb-kick-expire-test", clk) + + roomID := "room-hb-kick-expire" + ownerID := int64(7901) + viewerID := int64(8001) + createHeartbeatRecoveryRoom(t, svc, roomID, ownerID, "hb-kick-expire") + if _, err := svc.JoinRoom(ctx, &roomv1.JoinRoomRequest{ + Meta: roomservice.NewRequestMeta(roomID, viewerID), + Role: "audience", + }); err != nil { + t.Fatalf("join viewer failed: %v", err) + } + if _, err := svc.KickUser(ctx, &roomv1.KickUserRequest{ + Meta: roomservice.NewRequestMeta(roomID, ownerID), + TargetUserId: viewerID, + DurationMs: 1000, + }); err != nil { + t.Fatalf("duration kick failed: %v", err) + } + if _, err := svc.RoomHeartbeat(ctx, &roomv1.RoomHeartbeatRequest{ + Meta: roomservice.NewRequestMeta(roomID, viewerID), + }); err == nil || xerr.CodeOf(err) != xerr.PermissionDenied { + t.Fatalf("heartbeat inside ban window must stay PermissionDenied, got: %v", err) + } + + clk.now = clk.now.Add(2 * time.Second) + heartbeatResp, err := svc.RoomHeartbeat(ctx, &roomv1.RoomHeartbeatRequest{ + Meta: roomservice.NewRequestMeta(roomID, viewerID), + }) + if err != nil { + t.Fatalf("heartbeat after ban expiry must recover presence: %v", err) + } + if findRoomUser(heartbeatResp.GetRoom(), viewerID) == nil { + t.Fatalf("expired ban user must be back online: %+v", heartbeatResp.GetRoom()) + } +} + +func TestRoomHeartbeatDoesNotBypassRoomPassword(t *testing.T) { + ctx := context.Background() + repository := mysqltest.NewRepository(t) + svc := newHeartbeatRecoveryService(t, repository, "node-hb-password-test", nil) + + roomID := "room-hb-password" + ownerID := int64(8101) + viewerID := int64(8201) + createHeartbeatRecoveryRoom(t, svc, roomID, ownerID, "hb-password") + if _, err := svc.SetRoomPassword(ctx, &roomv1.SetRoomPasswordRequest{ + Meta: roomservice.NewRequestMeta(roomID, ownerID), + Locked: true, + Password: "4321", + }); err != nil { + t.Fatalf("set room password failed: %v", err) + } + if _, err := svc.JoinRoom(ctx, &roomv1.JoinRoomRequest{ + Meta: roomservice.NewRequestMeta(roomID, viewerID), + Role: "audience", + Password: "4321", + }); err != nil { + t.Fatalf("join locked room with password failed: %v", err) + } + if _, err := svc.LeaveRoom(ctx, &roomv1.LeaveRoomRequest{ + Meta: roomservice.NewRequestMeta(roomID, viewerID), + }); err != nil { + t.Fatalf("leave viewer failed: %v", err) + } + + // 心跳不携带密码，非房主不得借心跳绕过锁房，必须维持原 NotFound 契约走完整 JoinRoom。 + if _, err := svc.RoomHeartbeat(ctx, &roomv1.RoomHeartbeatRequest{ + Meta: roomservice.NewRequestMeta(roomID, viewerID), + }); err == nil || xerr.CodeOf(err) != xerr.NotFound { + t.Fatalf("locked room heartbeat recovery must stay NotFound for non-owner, got: %v", err) + } + + // 房主不受锁房密码约束（与 JoinRoom 一致），恢复后角色收敛为 owner。 + if _, err := svc.LeaveRoom(ctx, &roomv1.LeaveRoomRequest{ + Meta: roomservice.NewRequestMeta(roomID, ownerID), + }); err != nil { + t.Fatalf("leave owner failed: %v", err) + } + heartbeatResp, err := svc.RoomHeartbeat(ctx, &roomv1.RoomHeartbeatRequest{ + Meta: roomservice.NewRequestMeta(roomID, ownerID), + }) + if err != nil { + t.Fatalf("owner heartbeat recovery in locked room failed: %v", err) + } + owner := findRoomUser(heartbeatResp.GetRoom(), ownerID) + if owner == nil || owner.GetRole() != "owner" { + t.Fatalf("owner recovery must keep owner role: %+v", heartbeatResp.GetRoom()) + } +} diff --git a/services/room-service/internal/room/service/presence_replay_internal_test.go b/services/room-service/internal/room/service/presence_replay_internal_test.go new file mode 100644 index 00000000..12219461 --- /dev/null +++ b/services/room-service/internal/room/service/presence_replay_internal_test.go @@ -0,0 +1,74 @@ +package service + +import ( + "testing" + + "hyapp/services/room-service/internal/room/command" + "hyapp/services/room-service/internal/room/state" +) + +func replayHeartbeat(t *testing.T, current *state.RoomState, userID int64, sentAtMS int64) { + t.Helper() + cmd := &command.RoomHeartbeat{Base: command.Base{ + AppCode: "lalu", + RequestID: "req-replay-heartbeat", + CommandID: "cmd-replay-heartbeat", + ActorID: userID, + Room: current.RoomID, + SentAtMS: sentAtMS, + }} + if err := replay(current, cmd, sentAtMS); err != nil { + t.Fatalf("replay heartbeat for user %d failed: %v", userID, err) + } +} + +// 线上心跳会静默恢复丢失的 presence 后才提交命令日志；回放必须重现恢复（含 Version++）， +// 否则恢复出的 OnlineUsers/版本与快照谱系分叉。 +func TestReplayRoomHeartbeatRecreatesMissingPresence(t *testing.T) { + ownerID := int64(11) + adminID := int64(22) + viewerID := int64(33) + current := state.NewRoomState("room-replay-heartbeat", ownerID, 4, "voice") + current.AdminUsers[adminID] = true + baseVersion := current.Version + + replayHeartbeat(t, current, viewerID, 1_000) + viewer, exists := current.OnlineUsers[viewerID] + if !exists || viewer.Role != "audience" || viewer.JoinedAtMS != 1_000 || viewer.LastSeenAtMS != 1_000 { + t.Fatalf("replayed heal must recreate audience presence from SentAtMS: %+v", viewer) + } + if current.Version != baseVersion+1 { + t.Fatalf("replayed heal must advance version exactly once: base=%d now=%d", baseVersion, current.Version) + } + + // 角色从权威字段推导：owner/admin 恢复后徽标不丢失。 + replayHeartbeat(t, current, ownerID, 2_000) + if owner := current.OnlineUsers[ownerID]; owner == nil || owner.Role != "owner" { + t.Fatalf("replayed heal must restore owner role: %+v", owner) + } + replayHeartbeat(t, current, adminID, 3_000) + if admin := current.OnlineUsers[adminID]; admin == nil || admin.Role != "admin" { + t.Fatalf("replayed heal must restore admin role: %+v", admin) + } +} + +func TestReplayRoomHeartbeatRefreshesExistingPresence(t *testing.T) { + viewerID := int64(44) + current := state.NewRoomState("room-replay-heartbeat-refresh", 11, 4, "voice") + current.OnlineUsers[viewerID] = &state.RoomUserState{ + UserID: viewerID, + Role: "audience", + JoinedAtMS: 500, + LastSeenAtMS: 500, + } + baseVersion := current.Version + + replayHeartbeat(t, current, viewerID, 9_000) + viewer := current.OnlineUsers[viewerID] + if viewer.LastSeenAtMS != 9_000 || viewer.JoinedAtMS != 500 { + t.Fatalf("replayed refresh must only update last_seen: %+v", viewer) + } + if current.Version != baseVersion+1 { + t.Fatalf("replayed refresh must advance version exactly once: base=%d now=%d", baseVersion, current.Version) + } +} diff --git a/services/room-service/internal/room/service/recovery.go b/services/room-service/internal/room/service/recovery.go index 7845f32d..2911b773 100644 --- a/services/room-service/internal/room/service/recovery.go +++ b/services/room-service/internal/room/service/recovery.go @@ -187,8 +187,18 @@ func replay(current *state.RoomState, cmd command.Command, committedAtMS int64) case *command.RoomHeartbeat: existing, exists := current.OnlineUsers[typed.ActorUserID()] if !exists { - // 心跳和离房可能在同一房间版本附近并发落库；如果离房先被重放，随后到达的旧心跳 - // 不能重新创建 presence，也不能让整个 Room Cell 恢复失败。 + // 线上心跳对丢失的 presence 会执行静默恢复后才提交命令日志；回放遇到 + // "用户不存在"的已提交心跳，就是当时恢复成功的事实，必须重现恢复 + // （含 Version++），否则恢复出的 OnlineUsers/版本会与快照谱系分叉。 + // 守卫（room active/ban/锁房）在线上提交前已把关，被拒的心跳不会落库， + // 回放无需也不能重复判断——回放看到的 BanUsers 时点可能晚于命令时点。 + current.OnlineUsers[typed.ActorUserID()] = &state.RoomUserState{ + UserID: typed.ActorUserID(), + Role: recoveredPresenceRole(current, typed.ActorUserID()), + JoinedAtMS: typed.SentAtMS, + LastSeenAtMS: typed.SentAtMS, + } + current.Version++ return nil } // 心跳只刷新 last_seen，不产生用户进房事件。