diff --git a/services/room-service/internal/room/service/create_room.go b/services/room-service/internal/room/service/create_room.go index ef073fe0..6b5334b1 100644 --- a/services/room-service/internal/room/service/create_room.go +++ b/services/room-service/internal/room/service/create_room.go @@ -261,7 +261,7 @@ func (s *Service) CreateRoom(ctx context.Context, req *roomv1.CreateRoomRequest) projectionMS = elapsedMS(projectionStartedAt) // 持久化成功后再安装内存 Cell，避免内存有房间但恢复来源不完整。 - s.installCell(ctx, cmd.RoomID(), cell.New(roomState.Clone(), rank.FromState(roomState.GiftRank, s.rankLimit))) + s.installCell(ctx, cmd.RoomID(), cell.New(roomState.Clone(), rank.FromState(roomState.GiftRank, s.rankLimit)), lease.LeaseToken) logRoomCommandTiming(ctx, cmd.RoomID(), cmd.ID(), cmd.Type(), true, elapsedMS(startedAt), idempotencyMS, leaseMS, saveMutationMS, snapshotMS, projectionMS, 0, 0, tencentIMMS, true, true, slog.Int64("save_room_meta_ms", saveRoomMetaMS), ) diff --git a/services/room-service/internal/room/service/helpers.go b/services/room-service/internal/room/service/helpers.go index 0ed52f6d..779e8062 100644 --- a/services/room-service/internal/room/service/helpers.go +++ b/services/room-service/internal/room/service/helpers.go @@ -23,12 +23,12 @@ type loadedRoomRef struct { RoomID string } -func (s *Service) installCell(ctx context.Context, roomID string, roomCell *cell.RoomCell) { +func (s *Service) installCell(ctx context.Context, roomID string, roomCell *cell.RoomCell, leaseToken string) { s.mu.Lock() defer s.mu.Unlock() - // 注册表只保存当前进程内已装载 Cell，不代表 Redis lease 的全局 owner。 - s.cells[runtimeRoomKey(appcode.FromContext(ctx), roomID)] = roomCell + // 注册表只保存当前进程内已装载 Cell；leaseToken 决定它能否继续承载写路径。 + s.cells[runtimeRoomKey(appcode.FromContext(ctx), roomID)] = &loadedCell{roomCell: roomCell, leaseToken: leaseToken} } func (s *Service) loadCell(ctx context.Context, roomID string) *cell.RoomCell { @@ -36,7 +36,44 @@ func (s *Service) loadCell(ctx context.Context, roomID string) *cell.RoomCell { defer s.mu.Unlock() // 返回指针后由 RoomCell 自己的 mailbox 保证状态串行化。 - return s.cells[runtimeRoomKey(appcode.FromContext(ctx), roomID)] + loaded := s.cells[runtimeRoomKey(appcode.FromContext(ctx), roomID)] + if loaded == nil { + return nil + } + return loaded.roomCell +} + +func (s *Service) loadCellForLease(ctx context.Context, roomID string, lease router.Lease) *cell.RoomCell { + key := runtimeRoomKey(appcode.FromContext(ctx), roomID) + s.mu.Lock() + defer s.mu.Unlock() + + loaded := s.cells[key] + if loaded == nil { + return nil + } + if loaded.leaseToken == "" || loaded.leaseToken != lease.LeaseToken { + // lease token 变化代表当前节点经历过执行权断档；旧 Cell 可能落后于新 owner 已提交的 command log，必须丢弃后走恢复。 + delete(s.cells, key) + return nil + } + return loaded.roomCell +} + +func (s *Service) loadCellForCurrentOwner(ctx context.Context, roomID string) (*cell.RoomCell, error) { + if s.directory == nil { + // 测试中的极简 Service 可能没有装配 directory；生产 HealthCheck 会拒绝这种配置。 + return s.loadCell(ctx, roomID), nil + } + lease, exists, err := s.directory.Lookup(ctx, runtimeRoomKeyFromContext(ctx, roomID)) + if err != nil { + return nil, err + } + if !exists || lease.NodeID != s.nodeID || !lease.ValidAt(s.clock.Now()) { + // 只读路径不能主动接管 lease；没有当前 owner 身份时必须走持久化恢复，避免旧 Cell 放行过期状态。 + return nil, nil + } + return s.loadCellForLease(ctx, roomID, lease), nil } func (s *Service) forgetLoadedRoom(ctx context.Context, roomID string) { @@ -103,6 +140,10 @@ func (s *Service) loadedRoomLease(ctx context.Context, roomRef loadedRoomRef, no // 如果其他节点已经接管并持有有效 lease，当前节点必须跳过，避免旧内存态覆盖新 owner。 return router.Lease{}, false, nil } + if roomCell := s.loadCellForLease(ctx, roomRef.RoomID, lease); roomCell == nil { + // 旧 Cell 的 token 不连续时，后台 worker 不能靠重新抢到的 lease 继续写旧内存分支。 + return router.Lease{}, false, nil + } return lease, true, nil } diff --git a/services/room-service/internal/room/service/lease_cell_test.go b/services/room-service/internal/room/service/lease_cell_test.go new file mode 100644 index 00000000..e1bd9367 --- /dev/null +++ b/services/room-service/internal/room/service/lease_cell_test.go @@ -0,0 +1,99 @@ +package service + +import ( + "context" + "testing" + "time" + + "hyapp/pkg/appcode" + "hyapp/services/room-service/internal/room/cell" + "hyapp/services/room-service/internal/room/rank" + "hyapp/services/room-service/internal/room/state" + "hyapp/services/room-service/internal/router" +) + +func TestLoadedRoomLeaseDropsCellWhenLeaseTokenChanges(t *testing.T) { + ctx := appcode.WithContext(context.Background(), "lalu") + roomID := "room-token-gap" + now := time.Date(2026, 7, 7, 9, 0, 0, 0, time.UTC) + directory := router.NewMemoryDirectory() + svc := &Service{ + nodeID: "room-node-a", + leaseTTL: time.Second, + directory: directory, + cells: make(map[string]*loadedCell), + } + + lease, err := directory.EnsureOwner(ctx, runtimeRoomKey("lalu", roomID), svc.nodeID, now, svc.leaseTTL) + if err != nil { + t.Fatalf("ensure first lease: %v", err) + } + roomState := state.NewRoomState(roomID, 1001, 8, "voice") + roomState.Version = 7 + svc.installCell(ctx, roomID, cell.New(roomState, rank.FromState(roomState.GiftRank, 20)), lease.LeaseToken) + if got := svc.loadCellForLease(ctx, roomID, lease); got == nil { + t.Fatal("freshly installed cell should be reusable with the original lease token") + } + + if _, owned, err := svc.loadedRoomLease(ctx, loadedRoomRef{AppCode: "lalu", RoomID: roomID}, now.Add(2*time.Second)); err != nil { + t.Fatalf("load expired lease: %v", err) + } else if owned { + t.Fatal("expired lease re-acquisition must not keep using the old in-memory cell") + } + if got := svc.loadCell(ctx, roomID); got != nil { + t.Fatal("old in-memory cell must be forgotten after a lease-token discontinuity") + } + + currentLease, exists, err := directory.Lookup(ctx, runtimeRoomKey("lalu", roomID)) + if err != nil { + t.Fatalf("lookup current lease: %v", err) + } + if !exists || currentLease.LeaseToken == lease.LeaseToken { + t.Fatal("test setup expected the directory to issue a new lease token after expiry") + } +} + +func TestLoadCellForCurrentOwnerRejectsReadOnlyAndExpiredCells(t *testing.T) { + ctx := appcode.WithContext(context.Background(), "lalu") + roomID := "room-read-current-owner" + now := time.Date(2026, 7, 7, 10, 0, 0, 0, time.UTC) + directory := router.NewMemoryDirectory() + svc := &Service{ + nodeID: "room-node-a", + leaseTTL: time.Second, + clock: fixedRoomTestClock{now: now}, + directory: directory, + cells: make(map[string]*loadedCell), + } + roomState := state.NewRoomState(roomID, 1001, 8, "voice") + + lease, err := directory.EnsureOwner(ctx, runtimeRoomKey("lalu", roomID), svc.nodeID, now, svc.leaseTTL) + if err != nil { + t.Fatalf("ensure lease: %v", err) + } + svc.installCell(ctx, roomID, cell.New(roomState.Clone(), rank.FromState(roomState.GiftRank, 20)), "") + if got, err := svc.loadCellForCurrentOwner(ctx, roomID); err != nil { + t.Fatalf("load read-only token cell: %v", err) + } else if got != nil { + t.Fatal("read-only recovered cell must not be reused as current-owner state") + } + if got := svc.loadCell(ctx, roomID); got != nil { + t.Fatal("read-only cell should be forgotten after current-owner token check") + } + + svc.installCell(ctx, roomID, cell.New(roomState.Clone(), rank.FromState(roomState.GiftRank, 20)), lease.LeaseToken) + svc.clock = fixedRoomTestClock{now: now.Add(2 * time.Second)} + if got, err := svc.loadCellForCurrentOwner(ctx, roomID); err != nil { + t.Fatalf("load expired token cell: %v", err) + } else if got != nil { + t.Fatal("expired owner lease must not authorize reading the old in-memory cell") + } +} + +type fixedRoomTestClock struct { + now time.Time +} + +func (c fixedRoomTestClock) Now() time.Time { + return c.now +} diff --git a/services/room-service/internal/room/service/mic_publish_timeout.go b/services/room-service/internal/room/service/mic_publish_timeout.go index 25e23784..3e7ddc1c 100644 --- a/services/room-service/internal/room/service/mic_publish_timeout.go +++ b/services/room-service/internal/room/service/mic_publish_timeout.go @@ -48,7 +48,7 @@ func (s *Service) SweepMicPublishTimeouts(ctx context.Context) error { continue } - roomCell := s.loadCell(roomCtx, roomRef.RoomID) + roomCell := s.loadCellForLease(roomCtx, roomRef.RoomID, lease) if roomCell == nil { // loadedRoomRefs 和 loadCell 之间允许并发变化，缺失时跳过即可。 continue diff --git a/services/room-service/internal/room/service/pipeline.go b/services/room-service/internal/room/service/pipeline.go index f821f993..2239acdb 100644 --- a/services/room-service/internal/room/service/pipeline.go +++ b/services/room-service/internal/room/service/pipeline.go @@ -3,6 +3,7 @@ package service import ( "bytes" "context" + "fmt" "time" "hyapp/pkg/appcode" @@ -74,7 +75,13 @@ func (s *Service) mutateRoom(ctx context.Context, cmd command.Command, replayabl } result.snapshot = snapshotWithApp(ctx, result.snapshot) - if result.snapshot.GetVersion() != current.Version { + stateChanged := nextState.Version != current.Version + if stateChanged && nextState.Version != current.Version+1 { + // Room Cell 命令日志按每次状态变更推进一个版本恢复；跳版本或复用旧版本都说明命令实现破坏了恢复语义。 + return nil, xerr.New(xerr.Internal, fmt.Sprintf("room version advanced from %d to %d", current.Version, nextState.Version)) + } + + if stateChanged { // 版本变化代表本命令产生实际状态变更，需要持久化命令和事件。 if err := s.ensureLeaseStillOwned(ctx, cmd.RoomID(), lease); err != nil { return nil, err diff --git a/services/room-service/internal/room/service/presence.go b/services/room-service/internal/room/service/presence.go index 18ddfe1f..60ea3c86 100644 --- a/services/room-service/internal/room/service/presence.go +++ b/services/room-service/internal/room/service/presence.go @@ -461,7 +461,7 @@ func (s *Service) SweepStalePresence(ctx context.Context) error { continue } - roomCell := s.loadCell(roomCtx, roomRef.RoomID) + roomCell := s.loadCellForLease(roomCtx, roomRef.RoomID, lease) if roomCell == nil { continue } diff --git a/services/room-service/internal/room/service/recovery.go b/services/room-service/internal/room/service/recovery.go index 1892b928..a8f1419d 100644 --- a/services/room-service/internal/room/service/recovery.go +++ b/services/room-service/internal/room/service/recovery.go @@ -27,8 +27,8 @@ func (s *Service) ensureCell(ctx context.Context, roomID string) (*cell.RoomCell return nil, router.Lease{}, xerr.New(xerr.Conflict, fmt.Sprintf("room is owned by %s", lease.NodeID)) } - if roomCell := s.loadCell(ctx, roomID); roomCell != nil { - // 本节点已经装载该房间，直接复用已有 Room Cell。 + if roomCell := s.loadCellForLease(ctx, roomID, lease); roomCell != nil { + // 本节点已经装载该房间且 lease token 连续，才能直接复用已有 Room Cell。 return roomCell, lease, nil } @@ -55,7 +55,7 @@ func (s *Service) ensureCell(ctx context.Context, roomID string) (*cell.RoomCell // 恢复完成后创建新的 Room Cell，并从恢复后的 GiftRank 重建 LocalRank 索引。 roomCell := cell.New(restoredState, rank.FromState(restoredState.GiftRank, s.rankLimit)) - s.installCell(ctx, roomID, roomCell) + s.installCell(ctx, roomID, roomCell, lease.LeaseToken) return roomCell, lease, nil } diff --git a/services/room-service/internal/room/service/room_rocket.go b/services/room-service/internal/room/service/room_rocket.go index 696b9803..6ca2c0d4 100644 --- a/services/room-service/internal/room/service/room_rocket.go +++ b/services/room-service/internal/room/service/room_rocket.go @@ -834,7 +834,7 @@ func (s *Service) SweepRoomRocketLaunchings(ctx context.Context) error { if !owned { continue } - roomCell := s.loadCell(roomCtx, roomRef.RoomID) + roomCell := s.loadCellForLease(roomCtx, roomRef.RoomID, lease) if roomCell == nil { continue } diff --git a/services/room-service/internal/room/service/service.go b/services/room-service/internal/room/service/service.go index 69c7f0b5..f6b3f7f2 100644 --- a/services/room-service/internal/room/service/service.go +++ b/services/room-service/internal/room/service/service.go @@ -110,8 +110,8 @@ type Service struct { // mu 保护本进程内 room_id -> RoomCell 注册表。 mu sync.Mutex - // cells 保存当前节点已经装载的单房间执行单元。 - cells map[string]*cell.RoomCell + // cells 保存当前节点已经装载的单房间执行单元；写路径必须同时校验 lease token，避免 lease 断档后复用旧内存分支。 + cells map[string]*loadedCell // robotRuntimeMu 保护后台机器人房间运行循环；配置变更时要取消旧循环后重建。 robotRuntimeMu sync.Mutex @@ -134,6 +134,13 @@ type robotRoomRuntime struct { configUpdatedAtMS int64 } +type loadedCell struct { + // roomCell 是当前进程内的 Room Cell 状态副本，只有 leaseToken 连续匹配时才能继续承载写命令。 + roomCell *cell.RoomCell + // leaseToken 记录装载该 Cell 时持有的 Redis lease 实例；空值只允许读路径使用，不能参与写路径复用。 + leaseToken string +} + // mutationResult 是命令在 Room Cell 内执行后的统一结果包。 type mutationResult struct { // applied 表示本次命令确实改变并提交了房间状态。 @@ -272,7 +279,7 @@ func New(cfg Config, directory router.Directory, repository Repository, wallet i luckyGiftSendLocker: cfg.LuckyGiftSendLocker, luckyGiftSendLockTTL: luckyGiftSendLockTTL, rtcUserRemover: cfg.RTCUserRemover, - cells: make(map[string]*cell.RoomCell), + cells: make(map[string]*loadedCell), robotRuntimes: make(map[string]robotRoomRuntime), humanRobotRuntimes: make(map[string]robotRoomRuntime), } diff --git a/services/room-service/internal/room/service/snapshot.go b/services/room-service/internal/room/service/snapshot.go index 40a6a3c5..5bf700e1 100644 --- a/services/room-service/internal/room/service/snapshot.go +++ b/services/room-service/internal/room/service/snapshot.go @@ -6,14 +6,14 @@ import ( "google.golang.org/protobuf/proto" roomv1 "hyapp.local/api/proto/room/v1" "hyapp/pkg/appcode" - "hyapp/services/room-service/internal/room/cell" - "hyapp/services/room-service/internal/room/rank" ) // currentSnapshot 返回当前房间的最新快照；没有内存 Cell 时会从持久化恢复后再返回。 func (s *Service) currentSnapshot(ctx context.Context, roomID string) (*roomv1.RoomSnapshot, error) { - if roomCell := s.loadCell(ctx, roomID); roomCell != nil { - // 内存 Cell 是最新状态来源，guard 查询优先读它。 + if roomCell, err := s.loadCellForCurrentOwner(ctx, roomID); err != nil { + return nil, err + } else if roomCell != nil { + // 只有当前节点仍持有匹配 lease token 时，内存 Cell 才是最新状态来源。 currentState, _, err := roomCell.Snapshot(ctx) if err != nil { return nil, err @@ -38,9 +38,6 @@ func (s *Service) currentSnapshot(ctx context.Context, roomID string) (*roomv1.R return snapshotWithApp(ctx, restoredState.ToProto()), nil } - roomCell := cell.New(restoredState, rank.FromState(restoredState.GiftRank, s.rankLimit)) - s.installCell(ctx, roomID, roomCell) - return snapshotWithApp(ctx, restoredState.ToProto()), nil } diff --git a/services/room-service/internal/storage/mysql/command_log.go b/services/room-service/internal/storage/mysql/command_log.go index 17ca8058..c39eb120 100644 --- a/services/room-service/internal/storage/mysql/command_log.go +++ b/services/room-service/internal/storage/mysql/command_log.go @@ -5,6 +5,7 @@ import ( "context" "database/sql" "errors" + "fmt" "strings" "google.golang.org/protobuf/proto" @@ -70,6 +71,35 @@ func (r *Repository) SaveMutation(ctx context.Context, commit roomservice.Mutati return err } + if err := r.lockRoomMutation(ctx, tx, appCode, commit.Command.RoomID); err != nil { + _ = tx.Rollback() + return err + } + + if commit.Command.Replayable { + var existingCommandID string + err := tx.QueryRowContext(ctx, + `SELECT command_id + FROM room_command_log FORCE INDEX (idx_room_command_replay) + WHERE app_code = ? AND room_id = ? AND room_version = ? AND replayable = TRUE AND command_id <> ? + LIMIT 1 + FOR UPDATE`, + appCode, + commit.Command.RoomID, + commit.Command.RoomVersion, + commit.Command.CommandID, + ).Scan(&existingCommandID) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + _ = tx.Rollback() + return err + } + if existingCommandID != "" { + _ = tx.Rollback() + // 同房间同版本只能有一个可回放状态变更；这是恢复链路的最后一道持久化防线。 + return fmt.Errorf("room command version conflict: room_id=%s room_version=%d existing_command_id=%s new_command_id=%s", commit.Command.RoomID, commit.Command.RoomVersion, existingCommandID, commit.Command.CommandID) + } + } + if _, err := tx.ExecContext(ctx, `INSERT INTO room_command_log (app_code, room_id, room_version, command_id, command_type, owner_node_id, lease_token, payload, replayable, created_at_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) @@ -317,6 +347,26 @@ func (r *Repository) SaveMutation(ctx context.Context, commit roomservice.Mutati return tx.Commit() } +func (r *Repository) lockRoomMutation(ctx context.Context, tx *sql.Tx, appCode string, roomID string) error { + var lockedRoomID string + err := tx.QueryRowContext(ctx, + `SELECT room_id + FROM rooms + WHERE app_code = ? AND room_id = ? + FOR UPDATE`, + appCode, + roomID, + ).Scan(&lockedRoomID) + if errors.Is(err, sql.ErrNoRows) { + // CreateRoom 的测试级别 command-log 写入可能没有预置 rooms 行；生产命令在 SaveMutation 前已有 room meta。 + return nil + } + if err != nil { + return err + } + return nil +} + func (r *Repository) deleteRoomRows(ctx context.Context, tx *sql.Tx, appCode string, roomID string) error { // 删除覆盖所有能让房间重新出现在列表、恢复链路或 owner 唯一约束里的本服务表。 // room_outbox 也一起清理，避免已删除房间继续补偿投递旧的展示事件。 @@ -418,16 +468,16 @@ func (r *Repository) ListCommandsAfter(ctx context.Context, roomID string, roomV // SaveSnapshot 保存房间最新快照。 func (r *Repository) SaveSnapshot(ctx context.Context, snapshot roomservice.SnapshotRecord) error { - // 快照按 room_id 保留最新一条，版本更旧的写入不会覆盖较新的恢复来源。 + // 快照按 room_id 保留最新一条；同版本后写可能来自旧 owner 分支，不能覆盖已经存在的恢复来源。 appCode := normalizedRecordAppCode(ctx, snapshot.AppCode) _, err := r.db.ExecContext(ctx, `INSERT INTO room_snapshots (app_code, room_id, room_version, payload, created_at_ms, updated_at_ms) VALUES (?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE - room_version = IF(VALUES(room_version) >= room_version, VALUES(room_version), room_version), - payload = IF(VALUES(room_version) >= room_version, VALUES(payload), payload), - created_at_ms = IF(VALUES(room_version) >= room_version, VALUES(created_at_ms), created_at_ms), - updated_at_ms = IF(VALUES(room_version) >= room_version, VALUES(updated_at_ms), updated_at_ms)`, + payload = IF(VALUES(room_version) > room_version, VALUES(payload), payload), + created_at_ms = IF(VALUES(room_version) > room_version, VALUES(created_at_ms), created_at_ms), + updated_at_ms = IF(VALUES(room_version) > room_version, VALUES(updated_at_ms), updated_at_ms), + room_version = IF(VALUES(room_version) > room_version, VALUES(room_version), room_version)`, appCode, snapshot.RoomID, snapshot.RoomVersion, diff --git a/services/room-service/internal/storage/mysql/command_log_test.go b/services/room-service/internal/storage/mysql/command_log_test.go new file mode 100644 index 00000000..938ccebd --- /dev/null +++ b/services/room-service/internal/storage/mysql/command_log_test.go @@ -0,0 +1,196 @@ +package mysql + +import ( + "bytes" + "context" + "sync" + "testing" + + "hyapp/internal/testutil/mysqlschema" + "hyapp/pkg/appcode" + roomservice "hyapp/services/room-service/internal/room/service" +) + +func TestSaveSnapshotDoesNotOverwriteSameVersion(t *testing.T) { + ctx := appcode.WithContext(context.Background(), appcode.Default) + repo := newCommandLogTestRepository(t, ctx) + + goodPayload := []byte("snapshot-good-v10") + if err := repo.SaveSnapshot(ctx, roomservice.SnapshotRecord{ + AppCode: appcode.Default, + RoomID: "room-snapshot-version", + RoomVersion: 10, + Payload: goodPayload, + CreatedAtMS: 1000, + }); err != nil { + t.Fatalf("save initial snapshot: %v", err) + } + if err := repo.SaveSnapshot(ctx, roomservice.SnapshotRecord{ + AppCode: appcode.Default, + RoomID: "room-snapshot-version", + RoomVersion: 10, + Payload: []byte("snapshot-bad-same-version"), + CreatedAtMS: 2000, + }); err != nil { + t.Fatalf("save same-version snapshot: %v", err) + } + got, exists, err := repo.GetLatestSnapshot(ctx, "room-snapshot-version") + if err != nil { + t.Fatalf("get same-version snapshot: %v", err) + } + if !exists || got.RoomVersion != 10 || !bytes.Equal(got.Payload, goodPayload) || got.CreatedAtMS != 1000 { + t.Fatalf("same-version snapshot overwrote recovery source: got version=%d created=%d payload=%q", got.RoomVersion, got.CreatedAtMS, got.Payload) + } + + if err := repo.SaveSnapshot(ctx, roomservice.SnapshotRecord{ + AppCode: appcode.Default, + RoomID: "room-snapshot-version", + RoomVersion: 11, + Payload: []byte("snapshot-new-v11"), + CreatedAtMS: 3000, + }); err != nil { + t.Fatalf("save newer snapshot: %v", err) + } + got, exists, err = repo.GetLatestSnapshot(ctx, "room-snapshot-version") + if err != nil { + t.Fatalf("get newer snapshot: %v", err) + } + if !exists || got.RoomVersion != 11 || !bytes.Equal(got.Payload, []byte("snapshot-new-v11")) { + t.Fatalf("newer snapshot should replace old one: got version=%d payload=%q", got.RoomVersion, got.Payload) + } +} + +func TestSaveMutationRejectsDuplicateReplayableRoomVersion(t *testing.T) { + ctx := appcode.WithContext(context.Background(), appcode.Default) + repo := newCommandLogTestRepository(t, ctx) + roomID := "room-command-version" + + if err := repo.SaveMutation(ctx, roomservice.MutationCommit{Command: roomservice.CommandRecord{ + AppCode: appcode.Default, + RoomID: roomID, + RoomVersion: 20, + CommandID: "cmd-first", + CommandType: "join_room", + Payload: []byte(`{"command_id":"cmd-first"}`), + Replayable: true, + OwnerNodeID: "node-a", + LeaseToken: "lease-a", + CreatedAtMS: 1000, + }}); err != nil { + t.Fatalf("save first replayable command: %v", err) + } + if err := repo.SaveMutation(ctx, roomservice.MutationCommit{Command: roomservice.CommandRecord{ + AppCode: appcode.Default, + RoomID: roomID, + RoomVersion: 20, + CommandID: "cmd-second", + CommandType: "leave_room", + Payload: []byte(`{"command_id":"cmd-second"}`), + Replayable: true, + OwnerNodeID: "node-b", + LeaseToken: "lease-b", + CreatedAtMS: 2000, + }}); err == nil { + t.Fatal("duplicate replayable command version should be rejected") + } + if err := repo.SaveMutation(ctx, roomservice.MutationCommit{Command: roomservice.CommandRecord{ + AppCode: appcode.Default, + RoomID: roomID, + RoomVersion: 20, + CommandID: "cmd-noop", + CommandType: "join_room", + Payload: []byte(`{"command_id":"cmd-noop"}`), + Replayable: false, + OwnerNodeID: "node-a", + LeaseToken: "lease-a", + CreatedAtMS: 3000, + }}); err != nil { + t.Fatalf("same-version non-replayable idempotency record should still be saved: %v", err) + } +} + +func TestSaveMutationSerializesConcurrentReplayableRoomVersion(t *testing.T) { + ctx := appcode.WithContext(context.Background(), appcode.Default) + repo := newCommandLogTestRepository(t, ctx) + roomID := "room-command-version-race" + insertCommandLogTestRoom(t, ctx, repo, roomID) + + start := make(chan struct{}) + errs := make(chan error, 2) + var wg sync.WaitGroup + for _, commandID := range []string{"cmd-race-a", "cmd-race-b"} { + wg.Add(1) + go func(commandID string) { + defer wg.Done() + <-start + errs <- repo.SaveMutation(ctx, roomservice.MutationCommit{Command: roomservice.CommandRecord{ + AppCode: appcode.Default, + RoomID: roomID, + RoomVersion: 30, + CommandID: commandID, + CommandType: "room_heartbeat", + Payload: []byte(`{"command_id":"` + commandID + `"}`), + Replayable: true, + OwnerNodeID: "node-a", + LeaseToken: "lease-a", + CreatedAtMS: 4000, + }}) + }(commandID) + } + close(start) + wg.Wait() + close(errs) + + successes := 0 + failures := 0 + for err := range errs { + if err == nil { + successes++ + continue + } + failures++ + } + if successes != 1 || failures != 1 { + t.Fatalf("concurrent same-version replayable writes should produce one success and one conflict, got successes=%d failures=%d", successes, failures) + } +} + +func newCommandLogTestRepository(t *testing.T, ctx context.Context) *Repository { + t.Helper() + + schema := mysqlschema.New(t, mysqlschema.Config{ + EnvVar: "ROOM_SERVICE_MYSQL_TEST_DSN", + InitDBPath: mysqlschema.InitDBPath(t, mysqlschema.CallerFile(t, 1), "..", "..", "..", "deploy", "mysql", "initdb", "001_room_service.sql"), + DatabasePrefix: "hyapp_room_command_test", + }) + repo, err := Open(ctx, schema.DSN, 4, 4) + if err != nil { + t.Fatalf("open repository: %v", err) + } + t.Cleanup(func() { + _ = repo.Close() + }) + return repo +} + +func insertCommandLogTestRoom(t *testing.T, ctx context.Context, repo *Repository, roomID string) { + t.Helper() + + if _, err := repo.db.ExecContext(ctx, + `INSERT INTO rooms ( + app_code, room_id, room_short_id, owner_user_id, seat_count, mode, status, visible_region_id, created_at_ms, updated_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + appcode.Default, + roomID, + roomID+"-short", + 1001, + 8, + "voice", + "active", + 10, + 1000, + 1000, + ); err != nil { + t.Fatalf("insert command log test room: %v", err) + } +}