diff --git a/services/room-service/internal/room/service/recovery.go b/services/room-service/internal/room/service/recovery.go index 02bbf930..f1f2dbb8 100644 --- a/services/room-service/internal/room/service/recovery.go +++ b/services/room-service/internal/room/service/recovery.go @@ -248,7 +248,12 @@ func replay(current *state.RoomState, cmd command.Command, committedAtMS int64) case *command.ConfirmMicPublishing: seat, exists := current.SeatByUser(typed.TargetUserID) if !exists { - return fmt.Errorf("confirm_mic_publishing replay target is not on seat: %d", typed.TargetUserID) + // 确认事件可能晚于下麦/离房落库；运行态对此按 no-op 处理，恢复也不能让旧 RTC/客户端事件卡死房间。 + return nil + } + if seat.MicSessionID != typed.MicSessionID || seat.PublishState != state.MicPublishPending { + // 只允许当前 pending mic_session 被确认；旧会话、重复确认或已被替换的会话都按过期事件跳过。 + return nil } index := current.SeatIndex(seat.SeatNo) acceptedEventTimeMS := confirmMicPublishingAcceptedEventTimeMS(typed) @@ -278,7 +283,11 @@ func replay(current *state.RoomState, cmd command.Command, committedAtMS int64) case rtcEventAudioStarted: seat, exists := current.SeatByUser(typed.TargetUserID) if !exists { - return fmt.Errorf("apply_rtc_event audio_started replay target is not on seat: %d", typed.TargetUserID) + return nil + } + if seat.PublishState != state.MicPublishPending || typed.EventTimeMS <= seat.LastPublishEventTimeMS || (seat.PublishDeadlineMS > 0 && typed.EventTimeMS > seat.PublishDeadlineMS) { + // RTC 开始发布只确认当前 pending 会话；乱序、重复或超时事件恢复时保持 no-op。 + return nil } index := current.SeatIndex(seat.SeatNo) current.MicSeats[index].PublishState = state.MicPublishPublishing @@ -287,7 +296,11 @@ func replay(current *state.RoomState, cmd command.Command, committedAtMS int64) case rtcEventAudioStopped: seat, exists := current.SeatByUser(typed.TargetUserID) if !exists { - return fmt.Errorf("apply_rtc_event audio_stopped replay target is not on seat: %d", typed.TargetUserID) + return nil + } + if seat.MicSessionID == "" || (seat.LastPublishEventTimeMS > 0 && typed.EventTimeMS < seat.LastPublishEventTimeMS) { + // 停止发布只能关闭当前或更新的媒体会话；旧 stop 事件不能释放后续重上麦会话。 + return nil } index := current.SeatIndex(seat.SeatNo) current.ClearMicSession(index) @@ -296,10 +309,12 @@ func replay(current *state.RoomState, cmd command.Command, committedAtMS int64) // room_exited 是 RTC 媒体频道退出事实，不是 App 业务离房事实。 // 实时路径只释放麦位，业务 presence 只能由 LeaveRoom、KickUser、CloseRoom 或 stale worker 清理； // replay 必须保持相同语义，否则无最新 snapshot 的恢复会把仍在房间页的用户误删。 - if seat, exists := current.SeatByUser(typed.TargetUserID); exists { - index := current.SeatIndex(seat.SeatNo) - current.ClearMicSession(index) + seat, exists := current.SeatByUser(typed.TargetUserID) + if !exists { + return nil } + index := current.SeatIndex(seat.SeatNo) + current.ClearMicSession(index) current.Version++ default: return fmt.Errorf("apply_rtc_event replay unsupported event type: %s", typed.EventType) diff --git a/services/room-service/internal/room/service/recovery_test.go b/services/room-service/internal/room/service/recovery_test.go index 7759ac4b..a012267f 100644 --- a/services/room-service/internal/room/service/recovery_test.go +++ b/services/room-service/internal/room/service/recovery_test.go @@ -41,3 +41,67 @@ func TestReplayMicHeartbeatSkipsStaleSession(t *testing.T) { t.Fatalf("stale heartbeat must not refresh pending/replaced session, version=%d seat=%+v user=%+v", current.Version, current.MicSeats[0], current.OnlineUsers[7201]) } } + +func TestReplayConfirmMicPublishingSkipsStaleMissingSeat(t *testing.T) { + current := state.NewRoomState("room-replay-stale-confirm", 7301, 8, "voice") + current.OnlineUsers[7301] = &state.RoomUserState{UserID: 7301, Role: "owner", JoinedAtMS: 1000, LastSeenAtMS: 1000} + + if err := replay(current, &command.ConfirmMicPublishing{ + Base: command.Base{ActorID: 7301, Room: current.RoomID, SentAtMS: 2000}, + TargetUserID: 7301, + MicSessionID: "mic-old", + RoomVersion: 1, + EventTimeMS: 1900, + }, 2000); err != nil { + t.Fatalf("stale confirm without seat must be skipped: %v", err) + } + if current.Version != 0 { + t.Fatalf("stale confirm must not mutate version, got %d", current.Version) + } +} + +func TestReplayConfirmMicPublishingSkipsStaleSession(t *testing.T) { + current := state.NewRoomState("room-replay-stale-confirm-session", 7401, 8, "voice") + current.OnlineUsers[7401] = &state.RoomUserState{UserID: 7401, Role: "owner", JoinedAtMS: 1000, LastSeenAtMS: 1000} + current.MicSeats[0].UserID = 7401 + current.MicSeats[0].PublishState = state.MicPublishPending + current.MicSeats[0].MicSessionID = "mic-new" + + if err := replay(current, &command.ConfirmMicPublishing{ + Base: command.Base{ActorID: 7401, Room: current.RoomID, SentAtMS: 2000}, + TargetUserID: 7401, + MicSessionID: "mic-old", + RoomVersion: 1, + EventTimeMS: 1900, + }, 2000); err != nil { + t.Fatalf("stale confirm for replaced session must be skipped: %v", err) + } + if current.Version != 0 || current.MicSeats[0].PublishState != state.MicPublishPending { + t.Fatalf("stale confirm must not publish replaced session, version=%d seat=%+v", current.Version, current.MicSeats[0]) + } +} + +func TestReplayApplyRTCEventSkipsStaleMissingSeat(t *testing.T) { + current := state.NewRoomState("room-replay-stale-rtc", 7501, 8, "voice") + current.OnlineUsers[7501] = &state.RoomUserState{UserID: 7501, Role: "owner", JoinedAtMS: 1000, LastSeenAtMS: 1000} + + if err := replay(current, &command.ApplyRTCEvent{ + Base: command.Base{ActorID: 7501, Room: current.RoomID, SentAtMS: 2000}, + TargetUserID: 7501, + EventType: rtcEventAudioStarted, + EventTimeMS: 1900, + }, 2000); err != nil { + t.Fatalf("stale rtc start without seat must be skipped: %v", err) + } + if err := replay(current, &command.ApplyRTCEvent{ + Base: command.Base{ActorID: 7501, Room: current.RoomID, SentAtMS: 2100}, + TargetUserID: 7501, + EventType: rtcEventAudioStopped, + EventTimeMS: 2050, + }, 2100); err != nil { + t.Fatalf("stale rtc stop without seat must be skipped: %v", err) + } + if current.Version != 0 { + t.Fatalf("stale rtc events must not mutate version, got %d", current.Version) + } +}