修复房间问题
This commit is contained in:
zhx
parent
b8cbb93183
commit
a328b2bfef
@ -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),
|
||||
)
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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
|
||||
}
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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),
|
||||
}
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
|
||||
@ -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,
|
||||
|
||||
196
services/room-service/internal/storage/mysql/command_log_test.go
Normal file
196
services/room-service/internal/storage/mysql/command_log_test.go
Normal file
@ -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)
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user