From d67213fca638b02fee0b7af53b6ede373e34ce10 Mon Sep 17 00:00:00 2001 From: zhx Date: Tue, 21 Jul 2026 12:47:27 +0800 Subject: [PATCH] fix(game): retry room RPS settlement side effects --- .../internal/service/roomrps/service.go | 34 +++++- .../internal/service/roomrps/service_test.go | 113 +++++++++++++++++- 2 files changed, 144 insertions(+), 3 deletions(-) diff --git a/services/game-service/internal/service/roomrps/service.go b/services/game-service/internal/service/roomrps/service.go index 9e3629af..aec211a6 100644 --- a/services/game-service/internal/service/roomrps/service.go +++ b/services/game-service/internal/service/roomrps/service.go @@ -581,14 +581,44 @@ func (s *Service) RetrySettlement(ctx context.Context, app string, challengeID s return nil, 0, xerr.New(xerr.Conflict, "room rps challenge is not settlement_failed") } nowMS := s.now().UnixMilli() + // 先把失败单改为结算中，再释放进程锁调用 wallet/room；这个中间态阻止 + // 并发后台重试同时发两条 finished IM。wallet command_id 和 room command_id + // 都由 challenge_id 稳定派生，所以未知结果的跨服务请求可以安全重放。 challenge.Status = StatusFinished - challenge.SettlementStatus = SettlementSettled + challenge.SettlementStatus = SettlementPending challenge.FailureReason = "" - challenge.SettledAtMs = nowMS challenge.UpdatedAtMs = nowMS cloned := cloneChallenge(challenge) s.mu.Unlock() + settled, err := s.settleChallenge(ctx, "room-rps-retry:"+challengeID, cloned) + if err != nil { + // 重试仍失败时必须回到可重试终态；不发 finished IM，也不能让 + // 查询端把未完成的跨服务结算当成成功。已成功的单边动作依赖稳定 + // command_id 在下次重试时原样重放，不做反向补偿以免双退款。 + s.markSettlementFailed(app, challengeID, err) + return nil, 0, err + } + + nowMS = s.now().UnixMilli() + s.mu.Lock() + current := s.challenges[challengeID] + if current == nil || current.GetAppCode() != app { + s.mu.Unlock() + return nil, 0, xerr.New(xerr.NotFound, "room rps challenge not found") + } + // settleChallenge 返回 wallet 幂等回执中的真实余额；只有 room 礼物也成功后 + // 才将内存快照收敛为 settled，保证 finished IM 不会描述一份缺失的火箭礼物。 + current.Status = StatusFinished + current.Initiator.BalanceAfter = settled.GetInitiator().GetBalanceAfter() + current.Challenger.BalanceAfter = settled.GetChallenger().GetBalanceAfter() + current.SettlementStatus = settled.GetSettlementStatus() + current.FailureReason = "" + current.SettledAtMs = nowMS + current.UpdatedAtMs = nowMS + cloned = cloneChallenge(current) + s.mu.Unlock() + if err := s.publishChallengeEvent(ctx, eventFinished, cloned); err != nil { return nil, 0, err } diff --git a/services/game-service/internal/service/roomrps/service_test.go b/services/game-service/internal/service/roomrps/service_test.go index 919e3493..ea550a95 100644 --- a/services/game-service/internal/service/roomrps/service_test.go +++ b/services/game-service/internal/service/roomrps/service_test.go @@ -10,6 +10,7 @@ import ( gamev1 "hyapp.local/api/proto/game/v1" roomv1 "hyapp.local/api/proto/room/v1" walletv1 "hyapp.local/api/proto/wallet/v1" + "hyapp/pkg/tencentim" "hyapp/pkg/xerr" ) @@ -368,6 +369,84 @@ func TestAcceptChallengeRecordsChallengerCoinBalance(t *testing.T) { } } +func TestRetrySettlementReplaysIdempotentWalletAndRoomCommands(t *testing.T) { + wallet := newFakeRoomRPSWallet(map[int64]int64{41: 1000, 42: 500}) + // 模拟 room 已落地 command_id，但响应在网络上丢失的未知结果；这是 + // 最容易因盲目重试造成双充能的边界，必须靠稳定 room command_id 收敛。 + room := &fakeRoomRPSRoom{failAfterApply: 1} + publisher := &fakeRoomRPSPublisher{} + svc := New(Config{}, nil, publisher, wallet, room) + svc.now = func() time.Time { return time.UnixMilli(1770000000000) } + + created, _, err := svc.CreateChallenge(context.Background(), &gamev1.CreateRoomRPSChallengeRequest{ + Meta: &gamev1.RequestMeta{AppCode: "lalu", RequestId: "req-create-retry"}, + UserId: 41, + RoomId: "room-retry", + GiftId: 10001, + Gesture: "rock", + }) + if err != nil { + t.Fatalf("CreateChallenge() error = %v", err) + } + + _, _, err = svc.AcceptChallenge(context.Background(), &gamev1.AcceptRoomRPSChallengeRequest{ + Meta: &gamev1.RequestMeta{AppCode: "lalu", RequestId: "req-accept-retry"}, + UserId: 42, + ChallengeId: created.GetChallengeId(), + Gesture: "paper", + }) + if !xerr.IsCode(err, xerr.Unavailable) { + t.Fatalf("AcceptChallenge() error = %v, want room Unavailable", err) + } + failed, _, err := svc.GetChallenge(context.Background(), "lalu", created.GetChallengeId()) + if err != nil { + t.Fatalf("GetChallenge() failed state error = %v", err) + } + if failed.GetStatus() != StatusSettlementFailed || failed.GetSettlementStatus() != SettlementFailed { + t.Fatalf("room unknown result must remain retryable: %+v", failed) + } + if wallet.balances[41] != 900 || wallet.balances[42] != 500 { + t.Fatalf("first settlement must debit loser and refund winner once: balances=%+v", wallet.balances) + } + if len(room.requests) != 1 || len(publisher.messages) != 1 { + t.Fatalf("failed settlement must not publish terminal IM: room_calls=%d messages=%v", len(room.requests), publisher.messageTypes()) + } + + retried, _, err := svc.RetrySettlement(context.Background(), "lalu", created.GetChallengeId()) + if err != nil { + t.Fatalf("RetrySettlement() error = %v", err) + } + if retried.GetStatus() != StatusFinished || retried.GetSettlementStatus() != SettlementSettled || retried.GetChallenger().GetBalanceAfter() != 500 { + t.Fatalf("retry did not converge settled snapshot: %+v", retried) + } + if wallet.balances[41] != 900 || wallet.balances[42] != 500 { + t.Fatalf("wallet replay must not refund winner twice: balances=%+v", wallet.balances) + } + if len(wallet.applies) != 4 || wallet.applies[2].GetCommandId() != wallet.applies[3].GetCommandId() { + t.Fatalf("retry must replay the same wallet refund command: applies=%+v", wallet.applies) + } + if len(room.requests) != 2 || room.requests[0].GetMeta().GetCommandId() != room.requests[1].GetMeta().GetCommandId() { + t.Fatalf("retry must replay the same room gift command: requests=%+v", room.requests) + } + roomCommandID := room.requests[0].GetMeta().GetCommandId() + if room.applied[roomCommandID] != 1 { + t.Fatalf("room unknown-result replay must apply one gift, command_id=%q applied=%d", roomCommandID, room.applied[roomCommandID]) + } + if got := publisher.messageTypes(); fmt.Sprint(got) != "[room_rps_challenge_created room_rps_finished]" { + t.Fatalf("retry must publish exactly one terminal IM after all owners settle, got %v", got) + } + + walletCalls := len(wallet.applies) + roomCalls := len(room.requests) + messageCount := len(publisher.messages) + if _, _, err := svc.RetrySettlement(context.Background(), "lalu", created.GetChallengeId()); !xerr.IsCode(err, xerr.Conflict) { + t.Fatalf("second RetrySettlement() error = %v, want Conflict", err) + } + if len(wallet.applies) != walletCalls || len(room.requests) != roomCalls || len(publisher.messages) != messageCount { + t.Fatalf("settled retry must have no wallet, room, or IM side effects") + } +} + func TestAcceptChallengeReturnsTakenWhenAnotherUserAlreadyAccepted(t *testing.T) { wallet := newFakeRoomRPSWallet(map[int64]int64{41: 1000, 42: 500, 43: 500}) svc := New(Config{}, nil, nil, wallet) @@ -596,14 +675,46 @@ type fakeRoomRPSWallet struct { } type fakeRoomRPSRoom struct { - requests []*roomv1.ApplyRoomRPSGiftRequest + requests []*roomv1.ApplyRoomRPSGiftRequest + applied map[string]int + failAfterApply int } func (f *fakeRoomRPSRoom) ApplyRoomRPSGift(_ context.Context, req *roomv1.ApplyRoomRPSGiftRequest, _ ...grpc.CallOption) (*roomv1.ApplyRoomRPSGiftResponse, error) { f.requests = append(f.requests, req) + if f.applied == nil { + f.applied = make(map[string]int) + } + commandID := req.GetMeta().GetCommandId() + if f.applied[commandID] == 0 { + f.applied[commandID]++ + } + if f.failAfterApply > 0 { + f.failAfterApply-- + return nil, xerr.New(xerr.Unavailable, "room response lost after apply") + } return &roomv1.ApplyRoomRPSGiftResponse{}, nil } +type fakeRoomRPSPublisher struct { + messages []tencentim.CustomGroupMessage +} + +func (f *fakeRoomRPSPublisher) EnsureGroup(context.Context, tencentim.GroupSpec) error { return nil } + +func (f *fakeRoomRPSPublisher) PublishGroupCustomMessage(_ context.Context, message tencentim.CustomGroupMessage) error { + f.messages = append(f.messages, message) + return nil +} + +func (f *fakeRoomRPSPublisher) messageTypes() []string { + types := make([]string, 0, len(f.messages)) + for _, message := range f.messages { + types = append(types, message.Desc) + } + return types +} + func newFakeRoomRPSWallet(balances map[int64]int64) *fakeRoomRPSWallet { return &fakeRoomRPSWallet{balances: balances, replays: make(map[string]*walletv1.ApplyGameCoinChangeResponse)} }