From 53388616740e0f9d997757a0cf3378f1c06e52d0 Mon Sep 17 00:00:00 2001 From: zhx Date: Fri, 10 Jul 2026 11:51:33 +0800 Subject: [PATCH] fix: bound activity transaction retries --- .../internal/service/task/service.go | 2 +- .../internal/service/task/service_test.go | 42 ++++++++++++------- .../storage/mysql/growth_repository.go | 9 ++-- .../mysql/growth_repository_retry_test.go | 2 +- 4 files changed, 34 insertions(+), 21 deletions(-) diff --git a/services/activity-service/internal/service/task/service.go b/services/activity-service/internal/service/task/service.go index 0845c6b3..958a2a54 100644 --- a/services/activity-service/internal/service/task/service.go +++ b/services/activity-service/internal/service/task/service.go @@ -217,7 +217,7 @@ func (s *Service) HandleRoomEvent(ctx context.Context, envelope *roomeventsv1.Ev } // coin_spent 是钱包真实扣费金额；旧 outbox 可能只有礼物面值，兜底用 gift_value 保持历史事件可重放。 coinSpent := gift.GetCoinSpent() - if coinSpent <= 0 { + if coinSpent == 0 { coinSpent = gift.GetGiftValue() } dimensions, err := roomGiftTaskDimensions(envelope, &gift) diff --git a/services/activity-service/internal/service/task/service_test.go b/services/activity-service/internal/service/task/service_test.go index 61c1600e..898fe891 100644 --- a/services/activity-service/internal/service/task/service_test.go +++ b/services/activity-service/internal/service/task/service_test.go @@ -137,22 +137,34 @@ func TestConsumeTaskEventStillRejectsZeroValue(t *testing.T) { // TestHandleRoomEventStillRejectsNegativeSpend 确认兼容分支只确认合法零值，不把负值上游事实降级成计数事件。 func TestHandleRoomEventStillRejectsNegativeSpend(t *testing.T) { - repository := newRecordingTaskRepository() - svc := taskservice.New(repository, &fakeWalletClient{}) - envelope := roomGiftEnvelope(t, "evt-negative-gift-value", &roomeventsv1.RoomGiftSent{ - SenderUserId: 10001, - TargetUserId: 20002, - GiftId: "broken-gift", - GiftCount: 1, - GiftValue: -1, - CoinSpent: 0, - }) - - if _, err := svc.HandleRoomEvent(context.Background(), envelope); !xerr.IsCode(err, xerr.InvalidArgument) { - t.Fatalf("negative spend error=%v, want INVALID_ARGUMENT", err) + tests := []struct { + name string + giftValue int64 + coinSpent int64 + }{ + {name: "negative coin spent does not fallback", giftValue: 100, coinSpent: -1}, + {name: "negative legacy gift value", giftValue: -1, coinSpent: 0}, } - if repository.calls != 0 || len(repository.events) != 0 { - t.Fatalf("negative spend must fail before repository: calls=%d events=%+v", repository.calls, repository.events) + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + repository := newRecordingTaskRepository() + svc := taskservice.New(repository, &fakeWalletClient{}) + envelope := roomGiftEnvelope(t, "evt-negative-gift-value", &roomeventsv1.RoomGiftSent{ + SenderUserId: 10001, + TargetUserId: 20002, + GiftId: "broken-gift", + GiftCount: 1, + GiftValue: test.giftValue, + CoinSpent: test.coinSpent, + }) + + if _, err := svc.HandleRoomEvent(context.Background(), envelope); !xerr.IsCode(err, xerr.InvalidArgument) { + t.Fatalf("negative spend error=%v, want INVALID_ARGUMENT", err) + } + if repository.calls != 0 || len(repository.events) != 0 { + t.Fatalf("negative spend must fail before repository: calls=%d events=%+v", repository.calls, repository.events) + } + }) } } diff --git a/services/activity-service/internal/storage/mysql/growth_repository.go b/services/activity-service/internal/storage/mysql/growth_repository.go index bfb3ff3a..b861b663 100644 --- a/services/activity-service/internal/storage/mysql/growth_repository.go +++ b/services/activity-service/internal/storage/mysql/growth_repository.go @@ -210,7 +210,7 @@ func (r *Repository) ConsumeLevelEvent(ctx context.Context, event domain.ValueEv if attempt == growthTransactionMaxAttempts-1 { break } - // MQ 会并发回放同一热门用户的礼物事件；本地短退避先吸收瞬时锁冲突，避免每次 1213 都升级为 broker 秒级重试。 + // MQ 会并发回放同一热门用户的礼物事件；本地短退避先吸收瞬时死锁冲突，避免每次 1213 都升级为 broker 秒级重试。 timer := time.NewTimer(growthTransactionRetryDelay(event.EventID, attempt)) select { case <-ctx.Done(): @@ -224,7 +224,7 @@ func (r *Repository) ConsumeLevelEvent(ctx context.Context, event domain.ValueEv return domain.EventResult{}, lastErr } -// consumeLevelEventOnce 只执行一次完整事务；调用方只会在 MySQL 已回滚的 1213/1205 上重新 BeginTx。 +// consumeLevelEventOnce 只执行一次完整事务；调用方只会在 MySQL 已回滚的 1213 上重新 BeginTx。 func (r *Repository) consumeLevelEventOnce(ctx context.Context, event domain.ValueEvent, nowMS int64) (domain.EventResult, error) { if r == nil || r.db == nil { return domain.EventResult{}, xerr.New(xerr.Unavailable, "mysql repository is not configured") @@ -688,8 +688,9 @@ func isRetryableGrowthTransactionError(err error) bool { if !errors.As(err, &mysqlErr) { return false } - // 1213 是死锁受害事务，1205 是锁等待超时；两者都保证当前语句/事务没有成功提交，允许整事务重建。 - return mysqlErr.Number == 1213 || mysqlErr.Number == 1205 + // 1213 会立即回滚死锁受害事务，适合短退避后本地重建。1205 可能已等待完整的 + // innodb_lock_wait_timeout；在无 deadline 的 MQ 回调里重试它会让唯一消费协程继续占用数分钟，应交给 broker 延迟重投。 + return mysqlErr.Number == 1213 } func growthTransactionRetryDelay(eventID string, attempt int) time.Duration { diff --git a/services/activity-service/internal/storage/mysql/growth_repository_retry_test.go b/services/activity-service/internal/storage/mysql/growth_repository_retry_test.go index 53082105..27e50e2d 100644 --- a/services/activity-service/internal/storage/mysql/growth_repository_retry_test.go +++ b/services/activity-service/internal/storage/mysql/growth_repository_retry_test.go @@ -20,7 +20,7 @@ func TestIsRetryableGrowthTransactionError(t *testing.T) { }{ {name: "deadlock", err: &mysqldriver.MySQLError{Number: 1213, Message: "deadlock"}, want: true}, {name: "wrapped deadlock", err: fmt.Errorf("consume growth event: %w", &mysqldriver.MySQLError{Number: 1213, Message: "deadlock"}), want: true}, - {name: "lock wait timeout", err: &mysqldriver.MySQLError{Number: 1205, Message: "lock wait timeout"}, want: true}, + {name: "lock wait timeout", err: &mysqldriver.MySQLError{Number: 1205, Message: "lock wait timeout"}, want: false}, {name: "duplicate key", err: &mysqldriver.MySQLError{Number: 1062, Message: "duplicate"}, want: false}, {name: "text only", err: errors.New("Error 1213: Deadlock found"), want: false}, {name: "context canceled", err: context.Canceled, want: false},