修复用户上麦心跳
This commit is contained in:
zhx
parent
b9603832fc
commit
8715b19f28
@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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())
|
||||
|
||||
@ -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())
|
||||
}
|
||||
}
|
||||
@ -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)
|
||||
}
|
||||
}
|
||||
@ -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，不产生用户进房事件。
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user