From add80f781e56f4744fc8c4a8191b062d2fefa028 Mon Sep 17 00:00:00 2001
From: wangbo <wangbolhr@163.com>
Date: Fri, 31 Jul 2026 03:55:22 +0800
Subject: [PATCH] =?UTF-8?q?perf(wallet):=20=E9=81=BF=E5=85=8D=E5=90=8C?=
 =?UTF-8?q?=E9=92=B1=E5=8C=85=E9=A2=84=E7=95=99=E5=8D=A0=E6=BB=A1=E6=95=B0?=
 =?UTF-8?q?=E6=8D=AE=E5=BA=93=E6=B1=A0?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

千并发验收的 pg_stat_activity 显示大量事务锁等待集中在同一验收钱包，等待行锁的请求持续占用 API Pool，导致 canceled acquire 和排队增长。

在每个 Store 内增加 256 槽钱包预留分片锁，同一钱包先在进程内串行再申请数据库连接；跨 API 节点仍由 PostgreSQL 行锁保证余额与幂等正确性。

验证：Go 全量测试、临时 PostgreSQL 并发账务集成测试、gofmt 和 git diff --check 通过。
---
 apps/api/internal/store/postgres.go           |  3 ++-
 apps/api/internal/store/wallet.go             | 20 +++++++++++++++
 .../internal/store/wallet_reservation_test.go | 25 +++++++++++++++++++
 3 files changed, 47 insertions(+), 1 deletion(-)

diff --git a/apps/api/internal/store/postgres.go b/apps/api/internal/store/postgres.go
index 7f31b48..e219f78 100644
--- a/apps/api/internal/store/postgres.go
+++ b/apps/api/internal/store/postgres.go
@@ -21,7 +21,8 @@ import (
 )
 
 type Store struct {
-	pool *pgxpool.Pool
+	pool                   *pgxpool.Pool
+	walletReservationLocks walletReservationLockSet
 }
 
 const (
diff --git a/apps/api/internal/store/wallet.go b/apps/api/internal/store/wallet.go
index 439ceff..24d2e6f 100644
--- a/apps/api/internal/store/wallet.go
+++ b/apps/api/internal/store/wallet.go
@@ -5,14 +5,30 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"hash/fnv"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
 	"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 {
 	ID              string         `json:"id"`
 	GatewayTenantID string         `json:"gatewayTenantId,omitempty"`
@@ -153,12 +169,16 @@ func (s *Store) ReserveTaskBilling(ctx context.Context, task GatewayTask, user *
 		pricingSnapshot = emptyObjectIfNil(pricingSnapshots[0])
 	}
 	if exactAmount := walletString(pricingSnapshot["reservationAmount"]); exactAmount != "" {
+		unlock := s.walletReservationLocks.lock(gatewayUserID)
+		defer unlock()
 		return s.reserveTaskBillingExact(ctx, task, gatewayUserID, exactAmount, pricingSnapshot)
 	}
 	amounts := walletBillingAmounts(billings)
 	if len(amounts) == 0 {
 		return nil, nil
 	}
+	unlock := s.walletReservationLocks.lock(gatewayUserID)
+	defer unlock()
 
 	reservations := make([]WalletBillingReservation, 0, len(amounts))
 	pricingSnapshotJSON, _ := json.Marshal(sanitizeJSONForStorage(pricingSnapshot))
diff --git a/apps/api/internal/store/wallet_reservation_test.go b/apps/api/internal/store/wallet_reservation_test.go
index b9ec265..5aadd46 100644
--- a/apps/api/internal/store/wallet_reservation_test.go
+++ b/apps/api/internal/store/wallet_reservation_test.go
@@ -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) {
 	t.Helper()
 	suffix := strconv.FormatInt(time.Now().UnixNano(), 10)