fix(game): retry room RPS settlement side effects
This commit is contained in:
zhx
parent
ec0fb81eac
commit
d67213fca6
@ -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
|
||||
}
|
||||
|
||||
@ -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)}
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user