perf(wallet): 避免同钱包预留占满数据库池
千并发验收的 pg_stat_activity 显示大量事务锁等待集中在同一验收钱包,等待行锁的请求持续占用 API Pool,导致 canceled acquire 和排队增长。 在每个 Store 内增加 256 槽钱包预留分片锁,同一钱包先在进程内串行再申请数据库连接;跨 API 节点仍由 PostgreSQL 行锁保证余额与幂等正确性。 验证:Go 全量测试、临时 PostgreSQL 并发账务集成测试、gofmt 和 git diff --check 通过。
This commit is contained in:
@@ -21,7 +21,8 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
type Store struct {
|
type Store struct {
|
||||||
pool *pgxpool.Pool
|
pool *pgxpool.Pool
|
||||||
|
walletReservationLocks walletReservationLockSet
|
||||||
}
|
}
|
||||||
|
|
||||||
const (
|
const (
|
||||||
|
|||||||
@@ -5,14 +5,30 @@ import (
|
|||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"hash/fnv"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
|
"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
|
||||||
"github.com/jackc/pgx/v5"
|
"github.com/jackc/pgx/v5"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
const walletReservationLockStripes = 256
|
||||||
|
|
||||||
|
type walletReservationLockSet struct {
|
||||||
|
stripes [walletReservationLockStripes]sync.Mutex
|
||||||
|
}
|
||||||
|
|
||||||
|
func (locks *walletReservationLockSet) lock(gatewayUserID string) func() {
|
||||||
|
hasher := fnv.New32a()
|
||||||
|
_, _ = hasher.Write([]byte(strings.TrimSpace(gatewayUserID)))
|
||||||
|
stripe := &locks.stripes[hasher.Sum32()%walletReservationLockStripes]
|
||||||
|
stripe.Lock()
|
||||||
|
return stripe.Unlock
|
||||||
|
}
|
||||||
|
|
||||||
type GatewayWalletAccount struct {
|
type GatewayWalletAccount struct {
|
||||||
ID string `json:"id"`
|
ID string `json:"id"`
|
||||||
GatewayTenantID string `json:"gatewayTenantId,omitempty"`
|
GatewayTenantID string `json:"gatewayTenantId,omitempty"`
|
||||||
@@ -153,12 +169,16 @@ func (s *Store) ReserveTaskBilling(ctx context.Context, task GatewayTask, user *
|
|||||||
pricingSnapshot = emptyObjectIfNil(pricingSnapshots[0])
|
pricingSnapshot = emptyObjectIfNil(pricingSnapshots[0])
|
||||||
}
|
}
|
||||||
if exactAmount := walletString(pricingSnapshot["reservationAmount"]); exactAmount != "" {
|
if exactAmount := walletString(pricingSnapshot["reservationAmount"]); exactAmount != "" {
|
||||||
|
unlock := s.walletReservationLocks.lock(gatewayUserID)
|
||||||
|
defer unlock()
|
||||||
return s.reserveTaskBillingExact(ctx, task, gatewayUserID, exactAmount, pricingSnapshot)
|
return s.reserveTaskBillingExact(ctx, task, gatewayUserID, exactAmount, pricingSnapshot)
|
||||||
}
|
}
|
||||||
amounts := walletBillingAmounts(billings)
|
amounts := walletBillingAmounts(billings)
|
||||||
if len(amounts) == 0 {
|
if len(amounts) == 0 {
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
|
unlock := s.walletReservationLocks.lock(gatewayUserID)
|
||||||
|
defer unlock()
|
||||||
|
|
||||||
reservations := make([]WalletBillingReservation, 0, len(amounts))
|
reservations := make([]WalletBillingReservation, 0, len(amounts))
|
||||||
pricingSnapshotJSON, _ := json.Marshal(sanitizeJSONForStorage(pricingSnapshot))
|
pricingSnapshotJSON, _ := json.Marshal(sanitizeJSONForStorage(pricingSnapshot))
|
||||||
|
|||||||
@@ -113,6 +113,31 @@ func TestReserveTaskBillingSerializesConcurrentWalletReservations(t *testing.T)
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestWalletReservationLockSetSerializesSameWallet(t *testing.T) {
|
||||||
|
var locks walletReservationLockSet
|
||||||
|
unlock := locks.lock("wallet-user")
|
||||||
|
acquired := make(chan struct{})
|
||||||
|
released := make(chan struct{})
|
||||||
|
go func() {
|
||||||
|
defer close(released)
|
||||||
|
unlockSecond := locks.lock("wallet-user")
|
||||||
|
close(acquired)
|
||||||
|
unlockSecond()
|
||||||
|
}()
|
||||||
|
select {
|
||||||
|
case <-acquired:
|
||||||
|
t.Fatal("same-wallet reservation lock was acquired concurrently")
|
||||||
|
case <-time.After(20 * time.Millisecond):
|
||||||
|
}
|
||||||
|
unlock()
|
||||||
|
select {
|
||||||
|
case <-acquired:
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("same-wallet reservation lock did not unblock")
|
||||||
|
}
|
||||||
|
<-released
|
||||||
|
}
|
||||||
|
|
||||||
func seedWalletReservationUser(t *testing.T, ctx context.Context, db *Store) (string, string) {
|
func seedWalletReservationUser(t *testing.T, ctx context.Context, db *Store) (string, string) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
suffix := strconv.FormatInt(time.Now().UnixNano(), 10)
|
suffix := strconv.FormatInt(time.Now().UnixNano(), 10)
|
||||||
|
|||||||
Reference in New Issue
Block a user