package service import ( "context" "errors" "sync" "sync/atomic" "testing" "time" roomv1 "hyapp.local/api/proto/room/v1" "hyapp/pkg/appcode" "hyapp/services/room-service/internal/integration" "hyapp/services/room-service/internal/room/cell" "hyapp/services/room-service/internal/room/command" "hyapp/services/room-service/internal/room/outbox" "hyapp/services/room-service/internal/room/rank" "hyapp/services/room-service/internal/room/state" "hyapp/services/room-service/internal/router" ) // pipelineRaceRepository embeds the wide production boundary so this unit test can focus on the // command-log contract. Any unexpected repository call still panics through the nil embedded interface. type pipelineRaceRepository struct { Repository mu sync.Mutex commands map[string]CommandRecord getCalls atomic.Int32 saveCalls atomic.Int32 thirdGet chan struct{} // loseCommitAck simulates MySQL committing the transaction and then losing the COMMIT response. loseCommitAck atomic.Bool } func (r *pipelineRaceRepository) GetCommand(_ context.Context, roomID string, commandID string) (CommandRecord, bool, error) { if r.getCalls.Add(1) == 3 && r.thirdGet != nil { close(r.thirdGet) } r.mu.Lock() defer r.mu.Unlock() record, exists := r.commands[roomID+"\x00"+commandID] return record, exists, nil } func (r *pipelineRaceRepository) SaveMutation(_ context.Context, commit MutationCommit) error { r.saveCalls.Add(1) r.mu.Lock() key := commit.Command.RoomID + "\x00" + commit.Command.CommandID if _, exists := r.commands[key]; exists { r.mu.Unlock() return ErrCommandAlreadyCommitted } record := commit.Command record.Payload = append([]byte(nil), record.Payload...) record.ResultPayload = append([]byte(nil), record.ResultPayload...) r.commands[key] = record r.mu.Unlock() if r.loseCommitAck.CompareAndSwap(true, false) { return errors.New("mysql commit acknowledgement lost") } return nil } func TestMutateRoomRechecksCommandInsideCellAndReplaysTypedResult(t *testing.T) { ctx := appcode.WithContext(context.Background(), appcode.Default) repository := &pipelineRaceRepository{commands: make(map[string]CommandRecord), thirdGet: make(chan struct{})} directory := router.NewMemoryDirectory() svc := New(Config{NodeID: "node-pipeline-v2", LeaseTTL: 10 * time.Second, SnapshotEveryN: 1}, directory, repository, nil, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher()) roomID := "room-pipeline-v2" initial := state.NewRoomState(roomID, 101, 8, "voice") initial.Version = 1 lease, err := directory.EnsureOwner(ctx, runtimeRoomKeyFromContext(ctx, roomID), svc.nodeID, svc.clock.Now(), svc.leaseTTL) if err != nil { t.Fatalf("acquire test lease: %v", err) } svc.installCell(ctx, roomID, cell.New(initial, rank.New(20)), lease.LeaseToken) cmd := command.SendGift{Base: command.Base{AppCode: appcode.Default, CommandID: "cmd-pipeline-v2", ActorID: 101, Room: roomID}, TargetType: "user", TargetUserID: 202, TargetUserIDs: []int64{202}, RequestedTargetUserIDs: []int64{202}, GiftID: "rose", GiftCount: 1} started := make(chan struct{}) release := make(chan struct{}) var mutateCalls atomic.Int32 mutate := func(_ time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) { if mutateCalls.Add(1) == 1 { close(started) <-release } current.Heat += 100 current.Version++ return mutationResult{ snapshot: current.ToProto(), deleteRoom: true, // 跳过与本测试无关的 snapshot/list projections；Cell 本身仍正常提交。 giftV2: &roomv1.SendGiftResultV2{ ApiVersion: 2, CommandId: cmd.ID(), BillingReceiptId: "receipt-once", CoinBalanceAfter: 900, SettledGiftCount: 1, SettledTargetUserIds: []int64{202}, CommittedRoomVersion: current.Version, RoomHeat: current.Heat, OperationStatus: GiftOperationStatusCommitted, }, }, nil, nil } results := make(chan mutationResult, 2) errors := make(chan error, 2) go func() { result, err := svc.mutateRoom(ctx, cmd, true, mutate) results <- result errors <- err }() <-started go func() { result, err := svc.mutateRoom(ctx, cmd, true, mutate) results <- result errors <- err }() // 第三次读取是第二个请求的 Cell 外预检；确认它已经 miss 后再放开首请求，稳定复现历史竞态窗口。 <-repository.thirdGet close(release) replays := 0 for range 2 { if err := <-errors; err != nil { t.Fatalf("mutateRoom failed: %v", err) } result := <-results if result.giftV2 == nil || result.giftV2.GetBillingReceiptId() != "receipt-once" || result.giftV2.GetCoinBalanceAfter() != 900 { t.Fatalf("typed result was not preserved: %+v", result.giftV2) } if result.giftV2.GetIdempotentReplay() { replays++ } } if mutateCalls.Load() != 1 || replays != 1 { t.Fatalf("same command must mutate once and replay once: mutate_calls=%d replays=%d", mutateCalls.Load(), replays) } roomState, _, err := svc.cells[runtimeRoomKeyFromContext(ctx, roomID)].roomCell.Snapshot(ctx) if err != nil { t.Fatalf("read room state: %v", err) } if roomState.Heat != 100 || roomState.Version != 2 { t.Fatalf("duplicate command changed Cell twice: heat=%d version=%d", roomState.Heat, roomState.Version) } } func TestMutateRoomConvergesCellAfterCommittedTransactionLosesAck(t *testing.T) { ctx := appcode.WithContext(context.Background(), appcode.Default) repository := &pipelineRaceRepository{commands: make(map[string]CommandRecord)} repository.loseCommitAck.Store(true) directory := router.NewMemoryDirectory() svc := New(Config{NodeID: "node-pipeline-commit-unknown", LeaseTTL: 10 * time.Second, SnapshotEveryN: 1}, directory, repository, nil, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher()) roomID := "room-pipeline-commit-unknown" initial := state.NewRoomState(roomID, 101, 8, "voice") initial.Version = 1 lease, err := directory.EnsureOwner(ctx, runtimeRoomKeyFromContext(ctx, roomID), svc.nodeID, svc.clock.Now(), svc.leaseTTL) if err != nil { t.Fatalf("acquire test lease: %v", err) } svc.installCell(ctx, roomID, cell.New(initial, rank.New(20)), lease.LeaseToken) cmd := command.SendGift{Base: command.Base{AppCode: appcode.Default, CommandID: "cmd-commit-unknown", ActorID: 101, Room: roomID}, TargetType: "user", TargetUserID: 202, TargetUserIDs: []int64{202}, RequestedTargetUserIDs: []int64{202}, GiftID: "rose", GiftCount: 1} var debitAndMutationCalls atomic.Int32 mutate := func(_ time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) { debitAndMutationCalls.Add(1) current.Heat += 100 current.Version++ return mutationResult{ snapshot: current.ToProto(), deleteRoom: true, // Isolate the Cell/command-log contract from snapshot and list projections. giftV2: &roomv1.SendGiftResultV2{ ApiVersion: 2, CommandId: cmd.ID(), BillingReceiptId: "receipt-commit-unknown", CoinBalanceAfter: 900, BalanceVersion: 17, SettledGiftCount: 1, SettledTargetUserIds: []int64{202}, CommittedRoomVersion: current.Version, RoomHeat: current.Heat, OperationStatus: GiftOperationStatusCommitted, }, }, nil, nil } first, err := svc.mutateRoom(ctx, cmd, true, mutate) if err != nil { t.Fatalf("lost COMMIT acknowledgement should converge from the stored command: %v", err) } if !first.applied || first.giftV2 == nil || first.giftV2.GetBalanceVersion() != 17 { t.Fatalf("first typed result was not retained after commit confirmation: %+v", first.giftV2) } replay, err := svc.mutateRoom(ctx, cmd, true, mutate) if err != nil { t.Fatalf("retry should replay the confirmed typed result: %v", err) } if replay.giftV2 == nil || !replay.giftV2.GetIdempotentReplay() || replay.giftV2.GetBillingReceiptId() != "receipt-commit-unknown" || replay.giftV2.GetBalanceVersion() != 17 { t.Fatalf("typed replay drifted after unknown commit: %+v", replay.giftV2) } if debitAndMutationCalls.Load() != 1 || repository.saveCalls.Load() != 1 { t.Fatalf("unknown COMMIT must not debit or persist twice: mutate=%d save=%d", debitAndMutationCalls.Load(), repository.saveCalls.Load()) } roomState, _, err := svc.cells[runtimeRoomKeyFromContext(ctx, roomID)].roomCell.Snapshot(ctx) if err != nil { t.Fatalf("read converged room state: %v", err) } if roomState.Heat != 100 || roomState.Version != 2 { t.Fatalf("Cell did not converge exactly once: heat=%d version=%d", roomState.Heat, roomState.Version) } }