fix: bound activity transaction retries
This commit is contained in:
parent
ba8907f90b
commit
5338861674
@ -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)
|
||||
|
||||
@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -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},
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user