From dc20782918c4e177e70c4516504d9dd3ae192118 Mon Sep 17 00:00:00 2001 From: fengsilin Date: Thu, 9 Apr 2026 15:56:07 +0800 Subject: [PATCH] =?UTF-8?q?fix(region-sync):=20=E4=BF=AE=E5=A4=8D=20Settle?= =?UTF-8?q?/SyncManager=20=E7=AB=9E=E4=BA=89=E6=9D=A1=E4=BB=B6=E5=B9=B6?= =?UTF-8?q?=E4=BC=98=E5=8C=96=E5=B9=B6=E5=8F=91=E5=AE=89=E5=85=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 settled 状态,防止 SyncManager 在 PreConsume/Settle 之间同步未结算记录 - Settle 使用 AdjustSyncedQuota(增量 gorm.Expr)替代 UpdateSyncedQuota(绝对值),修复并发覆写 - 网络失败时标记所有记录为 failed 并增加 retry_count,避免无限重试 - 新增 ArchiveStalePendingRecords 定时清理崩溃请求的 pending 记录 - 简化 IsSyncedUser/IsLocalUser 移除冗余 id 参数 - Dockerfile 切换华为云 Go 镜像,移除不兼容的 GOEXPERIMENT=greenteagc Co-Authored-By: Claude --- Dockerfile | 3 +- common/region_sync.go | 33 ++----- common/region_sync_test.go | 59 ++---------- controller/region_sync.go | 2 +- model/pending_sync_record.go | 51 ++++++++-- model/region_sync_test.go | 115 ++++++++++++++++++++--- model/user.go | 12 ++- service/billing_session_sync_test.go | 5 +- service/region_sync/sync_manager.go | 11 ++- service/region_sync/sync_manager_test.go | 18 ++-- service/region_sync/sync_push.go | 2 +- service/synced_user_funding.go | 28 +++--- service/synced_user_funding_test.go | 52 +++++++++- 13 files changed, 255 insertions(+), 136 deletions(-) diff --git a/Dockerfile b/Dockerfile index c9be6ad..fd1ac58 100644 --- a/Dockerfile +++ b/Dockerfile @@ -11,14 +11,13 @@ COPY ./web . COPY ./VERSION . RUN DISABLE_ESLINT_PLUGIN='true' VITE_REACT_APP_VERSION=$(cat VERSION) bun run build -FROM golang:alpine AS builder2 +FROM swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/library/golang:1.26.1-alpine AS builder2 ENV GO111MODULE=on CGO_ENABLED=0 ENV GOPROXY=https://goproxy.cn,direct ARG TARGETOS ARG TARGETARCH ENV GOOS=${TARGETOS:-linux} GOARCH=${TARGETARCH:-amd64} -ENV GOEXPERIMENT=greenteagc WORKDIR /build diff --git a/common/region_sync.go b/common/region_sync.go index df68b78..9b1efb2 100644 --- a/common/region_sync.go +++ b/common/region_sync.go @@ -1,33 +1,16 @@ package common -import "errors" - const ( - ForeignUserIDStart = 10000000 - UserSourceLocal = "local" - UserSourceSynced = "synced" + UserSourceLocal = "local" + UserSourceSynced = "synced" ) -func IsSyncedUser(source string, id int) bool { - return source == UserSourceSynced && id > 0 && id < ForeignUserIDStart -} - -func IsLocalUser(source string, id int) bool { - if source == UserSourceLocal { - return true - } - if id >= ForeignUserIDStart { - return true - } - return false +// IsSyncedUser 通过 source 字段判断是否为同步用户 +func IsSyncedUser(source string) bool { + return source == UserSourceSynced } -func ValidateUserSource(source string, id int) error { - if source == UserSourceSynced && id >= ForeignUserIDStart { - return errors.New("synced 用户 ID 必须小于 1千万") - } - if source == UserSourceLocal && id > 0 && id < ForeignUserIDStart { - return errors.New("local 用户 ID 必须大于等于 1千万") - } - return nil +// IsLocalUser 通过 source 字段判断是否为本地用户(非 synced 即为 local) +func IsLocalUser(source string) bool { + return source != UserSourceSynced } diff --git a/common/region_sync_test.go b/common/region_sync_test.go index b6b1c07..e4ffd5e 100644 --- a/common/region_sync_test.go +++ b/common/region_sync_test.go @@ -4,12 +4,6 @@ import ( "testing" ) -func TestForeignUserIDStart(t *testing.T) { - if ForeignUserIDStart != 10000000 { - t.Errorf("ForeignUserIDStart = %d, want 10000000", ForeignUserIDStart) - } -} - func TestUserSourceConstants(t *testing.T) { if UserSourceLocal != "local" { t.Errorf("UserSourceLocal = %q, want %q", UserSourceLocal, "local") @@ -23,21 +17,17 @@ func TestIsSyncedUser(t *testing.T) { tests := []struct { name string source string - id int want bool }{ - {"synced with normal id", UserSourceSynced, 123, true}, - {"synced with id just below threshold", UserSourceSynced, ForeignUserIDStart - 1, true}, - {"synced with id 0", UserSourceSynced, 0, false}, - {"synced with id at threshold", UserSourceSynced, ForeignUserIDStart, false}, - {"local with normal id", UserSourceLocal, 123, false}, - {"empty source with normal id", "", 123, false}, + {"synced source", UserSourceSynced, true}, + {"local source", UserSourceLocal, false}, + {"empty source", "", false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if got := IsSyncedUser(tt.source, tt.id); got != tt.want { - t.Errorf("IsSyncedUser(%q, %d) = %v, want %v", tt.source, tt.id, got, tt.want) + if got := IsSyncedUser(tt.source); got != tt.want { + t.Errorf("IsSyncedUser(%q) = %v, want %v", tt.source, got, tt.want) } }) } @@ -47,46 +37,17 @@ func TestIsLocalUser(t *testing.T) { tests := []struct { name string source string - id int want bool }{ - {"local at threshold", UserSourceLocal, ForeignUserIDStart, true}, - {"local above threshold", UserSourceLocal, ForeignUserIDStart + 100, true}, - {"local with id 0", UserSourceLocal, 0, true}, - {"synced with small id", UserSourceSynced, 123, false}, - {"empty source at threshold", "", ForeignUserIDStart, true}, - {"empty source with small id", "", 500, false}, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - if got := IsLocalUser(tt.source, tt.id); got != tt.want { - t.Errorf("IsLocalUser(%q, %d) = %v, want %v", tt.source, tt.id, got, tt.want) - } - }) - } -} - -func TestValidateUserSource(t *testing.T) { - tests := []struct { - name string - source string - id int - wantErr bool - }{ - {"synced with valid id", UserSourceSynced, 100, false}, - {"synced at threshold", UserSourceSynced, ForeignUserIDStart, true}, - {"local at threshold", UserSourceLocal, ForeignUserIDStart, false}, - {"local with small non-zero id", UserSourceLocal, 500, true}, - {"local with id 0", UserSourceLocal, 0, false}, - {"empty source with id 0", "", 0, false}, + {"local source", UserSourceLocal, true}, + {"empty source", "", true}, + {"synced source", UserSourceSynced, false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - err := ValidateUserSource(tt.source, tt.id) - if (err != nil) != tt.wantErr { - t.Errorf("ValidateUserSource(%q, %d) error = %v, wantErr %v", tt.source, tt.id, err, tt.wantErr) + if got := IsLocalUser(tt.source); got != tt.want { + t.Errorf("IsLocalUser(%q) = %v, want %v", tt.source, got, tt.want) } }) } diff --git a/controller/region_sync.go b/controller/region_sync.go index 9d193ec..14d93e4 100644 --- a/controller/region_sync.go +++ b/controller/region_sync.go @@ -242,7 +242,7 @@ func processDeductRecordWithTx(c *gin.Context, tx *gorm.DB, record region_sync.B return result } - // 查询扣费后的余额 + // 查询扣费后的余额(同一事务内可读到自己的更新,MySQL REPEATABLE READ / SQLite 均安全) var user model.User if err := tx.Where("id = ?", record.UserId).First(&user).Error; err == nil { result.RemainingQuota = user.Quota diff --git a/model/pending_sync_record.go b/model/pending_sync_record.go index 3eb6180..587bcc5 100644 --- a/model/pending_sync_record.go +++ b/model/pending_sync_record.go @@ -8,6 +8,7 @@ import ( const ( PendingSyncStatusPending = "pending" + PendingSyncStatusSettled = "settled" PendingSyncStatusSynced = "synced" PendingSyncStatusFailed = "failed" PendingSyncStatusArchived = "archived" @@ -36,13 +37,13 @@ func (PendingSyncRecord) TableName() string { func GetPendingSyncQuota(userId int) int { var total int64 DB.Model(&PendingSyncRecord{}). - Where("user_id = ? AND status IN ?", userId, []string{PendingSyncStatusPending, PendingSyncStatusFailed}). + Where("user_id = ? AND status IN ?", userId, []string{PendingSyncStatusPending, PendingSyncStatusSettled, PendingSyncStatusFailed}). Select("COALESCE(SUM(quota), 0)"). Scan(&total) return int(total) } -// CreatePendingSyncRecord 创建待同步记录 +// CreatePendingSyncRecord 创建待同步记录(status=pending) func CreatePendingSyncRecord(userId, remoteUserId int, requestId string, quota, estimatedQuota int) error { record := PendingSyncRecord{ UserId: userId, @@ -56,11 +57,40 @@ func CreatePendingSyncRecord(userId, remoteUserId int, requestId string, quota, return DB.Create(&record).Error } -// GetPendingRecordsForSync 获取待同步记录 +// CreateSettledSyncRecord 创建已结算的同步记录(PreConsume 跳过时使用) +func CreateSettledSyncRecord(userId, remoteUserId int, requestId string, quota int) error { + record := PendingSyncRecord{ + UserId: userId, + RemoteUserId: remoteUserId, + RequestId: requestId, + Quota: quota, + EstimatedQuota: quota, + Status: PendingSyncStatusSettled, + CreatedAt: time.Now().Unix(), + } + return DB.Create(&record).Error +} + +// SettlePendingSyncRecord 将 pending 记录结算为 settled,同时更新 quota +// 返回 (是否找到并更新, 错误) +func SettlePendingSyncRecord(requestId string, actualQuota int) (bool, error) { + result := DB.Model(&PendingSyncRecord{}). + Where("request_id = ? AND status = ?", requestId, PendingSyncStatusPending). + Updates(map[string]interface{}{ + "quota": actualQuota, + "status": PendingSyncStatusSettled, + }) + if result.Error != nil { + return false, result.Error + } + return result.RowsAffected > 0, nil +} + +// GetPendingRecordsForSync 获取待同步记录(只同步 settled 和 failed) func GetPendingRecordsForSync(limit, maxRetry int) []PendingSyncRecord { var records []PendingSyncRecord DB.Where("status IN ? AND retry_count < ?", - []string{PendingSyncStatusPending, PendingSyncStatusFailed}, + []string{PendingSyncStatusSettled, PendingSyncStatusFailed}, maxRetry). Limit(limit). Order("created_at asc"). @@ -85,11 +115,13 @@ func MarkRecordFailed(recordId int, errMsg string) error { }).Error } -// UpdatePendingSyncRecordQuota 更新待同步记录的实际扣费额度,返回影响的行数 -func UpdatePendingSyncRecordQuota(requestId string, actualQuota int) int64 { + +// ArchiveStalePendingRecords 将超时的 pending 记录标记为 archived(请求崩溃的情况) +func ArchiveStalePendingRecords(timeoutSeconds int64) int64 { + cutoff := time.Now().Unix() - timeoutSeconds result := DB.Model(&PendingSyncRecord{}). - Where("request_id = ? AND status = ?", requestId, PendingSyncStatusPending). - Update("quota", actualQuota) + Where("status = ? AND created_at < ?", PendingSyncStatusPending, cutoff). + Update("status", PendingSyncStatusArchived) if result.Error != nil { return 0 } @@ -114,7 +146,7 @@ func GetPendingQuotaByUser() map[int]int { var results []result DB.Model(&PendingSyncRecord{}). Select("user_id, COALESCE(SUM(quota), 0) as total"). - Where("status IN ?", []string{PendingSyncStatusPending, PendingSyncStatusFailed}). + Where("status IN ?", []string{PendingSyncStatusPending, PendingSyncStatusSettled, PendingSyncStatusFailed}). Group("user_id"). Scan(&results) @@ -124,3 +156,4 @@ func GetPendingQuotaByUser() map[int]int { } return m } + diff --git a/model/region_sync_test.go b/model/region_sync_test.go index 1258c1f..d9a50ea 100644 --- a/model/region_sync_test.go +++ b/model/region_sync_test.go @@ -42,11 +42,11 @@ func TestUser_IsSyncedUser_Method(t *testing.T) { user User want bool }{ - {"synced valid", User{Source: "synced", Id: 100}, true}, - {"synced boundary", User{Source: "synced", Id: common.ForeignUserIDStart - 1}, true}, - {"synced id 0", User{Source: "synced", Id: 0}, false}, - {"synced too large", User{Source: "synced", Id: common.ForeignUserIDStart}, false}, + {"synced with normal id", User{Source: "synced", Id: 100}, true}, + {"synced with large id", User{Source: "synced", Id: 99999999}, true}, + {"synced with id 0", User{Source: "synced", Id: 0}, true}, {"local user", User{Source: "local", Id: 100}, false}, + {"empty source", User{Source: "", Id: 100}, false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -61,9 +61,9 @@ func TestUser_IsLocalUser_Method(t *testing.T) { user User want bool }{ - {"local source", User{Source: "local", Id: common.ForeignUserIDStart}, true}, - {"foreign id", User{Source: "", Id: common.ForeignUserIDStart + 1}, true}, - {"synced source", User{Source: "synced", Id: 100}, false}, + {"local source", User{Source: "local", Id: 100}, true}, + {"empty source is local", User{Source: "", Id: 100}, true}, + {"synced source is not local", User{Source: "synced", Id: 100}, false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -187,12 +187,16 @@ func TestGetPendingSyncQuota(t *testing.T) { require.NoError(t, CreatePendingSyncRecord(userId, 200, "req-004", 300, 100)) DB.Model(&PendingSyncRecord{}).Where("request_id = ?", "req-004").Update("status", PendingSyncStatusFailed) + // 创建一个 settled 状态的记录,应该被计入 + require.NoError(t, CreatePendingSyncRecord(userId, 200, "req-005", 400, 200)) + DB.Model(&PendingSyncRecord{}).Where("request_id = ?", "req-005").Update("status", PendingSyncStatusSettled) + // 创建一个 synced 状态的记录,不应该被计入 - require.NoError(t, CreatePendingSyncRecord(userId, 200, "req-005", 9999, 0)) - DB.Model(&PendingSyncRecord{}).Where("request_id = ?", "req-005").Update("status", PendingSyncStatusSynced) + require.NoError(t, CreatePendingSyncRecord(userId, 200, "req-006", 9999, 0)) + DB.Model(&PendingSyncRecord{}).Where("request_id = ?", "req-006").Update("status", PendingSyncStatusSynced) total := GetPendingSyncQuota(userId) - assert.Equal(t, 3800, total) // 1000 + 2000 + 500 + 300 = 3800 + assert.Equal(t, 4200, total) // 1000 + 2000 + 500 + 300 + 400 = 4200 } func TestGetPendingSyncQuota_Empty(t *testing.T) { @@ -259,21 +263,29 @@ func TestGetPendingRecordsForSync_Basic(t *testing.T) { db := setupPendingSyncDB(t) now := time.Now().Unix() + // settled 记录会被同步 require.NoError(t, db.Create(&PendingSyncRecord{ UserId: 100, RemoteUserId: 200, RequestId: "req-rs-001", - Quota: 100, Status: PendingSyncStatusPending, CreatedAt: now - 100, + Quota: 100, Status: PendingSyncStatusSettled, CreatedAt: now - 100, }).Error) + // failed 记录会被重试 require.NoError(t, db.Create(&PendingSyncRecord{ UserId: 101, RemoteUserId: 201, RequestId: "req-rs-002", Quota: 200, Status: PendingSyncStatusFailed, CreatedAt: now - 50, }).Error) + // synced 记录不会被查到 require.NoError(t, db.Create(&PendingSyncRecord{ UserId: 102, RemoteUserId: 202, RequestId: "req-rs-003", Quota: 300, Status: PendingSyncStatusSynced, CreatedAt: now, }).Error) + // pending 记录不会被同步(等待 Settle) + require.NoError(t, db.Create(&PendingSyncRecord{ + UserId: 103, RemoteUserId: 203, RequestId: "req-rs-004", + Quota: 400, Status: PendingSyncStatusPending, CreatedAt: now + 1, + }).Error) records := GetPendingRecordsForSync(10, 3) - assert.Len(t, records, 2) + assert.Len(t, records, 2) // 只查 settled + failed // 按 created_at 升序 assert.Equal(t, "req-rs-001", records[0].RequestId) assert.Equal(t, "req-rs-002", records[1].RequestId) @@ -289,7 +301,7 @@ func TestGetPendingRecordsForSync_MaxRetry(t *testing.T) { }).Error) require.NoError(t, db.Create(&PendingSyncRecord{ UserId: 101, RemoteUserId: 201, RequestId: "req-retry-002", - Quota: 200, Status: PendingSyncStatusPending, RetryCount: 2, CreatedAt: now, + Quota: 200, Status: PendingSyncStatusSettled, RetryCount: 2, CreatedAt: now, }).Error) records := GetPendingRecordsForSync(10, 3) @@ -304,7 +316,7 @@ func TestGetPendingRecordsForSync_Limit(t *testing.T) { for i := 0; i < 5; i++ { require.NoError(t, db.Create(&PendingSyncRecord{ UserId: 100, RemoteUserId: 200, RequestId: fmt.Sprintf("req-lim-%d", i), - Quota: 100, Status: PendingSyncStatusPending, CreatedAt: now + int64(i), + Quota: 100, Status: PendingSyncStatusSettled, CreatedAt: now + int64(i), }).Error) } @@ -402,18 +414,91 @@ func TestGetPendingQuotaByUser(t *testing.T) { UserId: 101, RemoteUserId: 201, RequestId: "req-pq-003", Quota: 500, Status: PendingSyncStatusPending, CreatedAt: now, }).Error) + // settled 记录应计入 require.NoError(t, db.Create(&PendingSyncRecord{ UserId: 100, RemoteUserId: 200, RequestId: "req-pq-004", + Quota: 150, Status: PendingSyncStatusSettled, CreatedAt: now, + }).Error) + // synced 记录不计入 + require.NoError(t, db.Create(&PendingSyncRecord{ + UserId: 100, RemoteUserId: 200, RequestId: "req-pq-005", Quota: 9999, Status: PendingSyncStatusSynced, CreatedAt: now, }).Error) m := GetPendingQuotaByUser() - assert.Equal(t, 300, m[100]) // 100 + 200, synced 不计入 + assert.Equal(t, 450, m[100]) // 100 + 200 + 150 (pending + failed + settled), synced 不计入 assert.Equal(t, 500, m[101]) _, exists := m[102] assert.False(t, exists) } + +// --------------------------------------------------------------------------- +// Settled status tests +// --------------------------------------------------------------------------- + +func TestSettlePendingSyncRecord(t *testing.T) { + db := setupPendingSyncDB(t) + + now := time.Now().Unix() + require.NoError(t, db.Create(&PendingSyncRecord{ + UserId: 100, RemoteUserId: 200, RequestId: "req-settle-1", + Quota: 1000, EstimatedQuota: 1000, Status: PendingSyncStatusPending, CreatedAt: now, + }).Error) + + found, err := SettlePendingSyncRecord("req-settle-1", 700) + require.NoError(t, err) + assert.True(t, found) + + var record PendingSyncRecord + require.NoError(t, db.Where("request_id = ?", "req-settle-1").First(&record).Error) + assert.Equal(t, PendingSyncStatusSettled, record.Status) + assert.Equal(t, 700, record.Quota) +} + +func TestSettlePendingSyncRecord_NotFound(t *testing.T) { + setupPendingSyncDB(t) + + found, err := SettlePendingSyncRecord("nonexistent", 100) + require.NoError(t, err) + assert.False(t, found) +} + +func TestCreateSettledSyncRecord(t *testing.T) { + db := setupPendingSyncDB(t) + + err := CreateSettledSyncRecord(100, 200, "req-settled-direct", 500) + require.NoError(t, err) + + var record PendingSyncRecord + require.NoError(t, db.Where("request_id = ?", "req-settled-direct").First(&record).Error) + assert.Equal(t, PendingSyncStatusSettled, record.Status) + assert.Equal(t, 500, record.Quota) +} + +func TestArchiveStalePendingRecords(t *testing.T) { + db := setupPendingSyncDB(t) + + now := time.Now().Unix() + // 超时的 pending 记录(应被归档) + require.NoError(t, db.Create(&PendingSyncRecord{ + UserId: 100, RemoteUserId: 200, RequestId: "req-stale-1", + Quota: 100, Status: PendingSyncStatusPending, CreatedAt: now - 600, + }).Error) + // 未超时的 pending 记录(不应被归档) + require.NoError(t, db.Create(&PendingSyncRecord{ + UserId: 100, RemoteUserId: 200, RequestId: "req-stale-2", + Quota: 200, Status: PendingSyncStatusPending, CreatedAt: now - 100, + }).Error) + + archived := ArchiveStalePendingRecords(300) // 5分钟超时 + assert.Equal(t, int64(1), archived) + + var record PendingSyncRecord + require.NoError(t, db.Where("request_id = ?", "req-stale-1").First(&record).Error) + assert.Equal(t, PendingSyncStatusArchived, record.Status) +} + // --------------------------------------------------------------------------- // QuotaUpdateCallback tests // --------------------------------------------------------------------------- diff --git a/model/user.go b/model/user.go index 6b43423..f56c4b9 100644 --- a/model/user.go +++ b/model/user.go @@ -69,12 +69,12 @@ type User struct { // IsSyncedUser 判断是否为国内同步用户 func (u *User) IsSyncedUser() bool { - return common.IsSyncedUser(u.Source, u.Id) + return common.IsSyncedUser(u.Source) } // IsLocalUser 判断是否为本地用户 func (u *User) IsLocalUser() bool { - return common.IsLocalUser(u.Source, u.Id) + return common.IsLocalUser(u.Source) } func (user *User) ToBaseUser() *UserBase { @@ -996,6 +996,7 @@ func UpdateSyncedQuota(userId int, quota int) error { } // AtomicDecreaseSyncedQuota 原子扣减 synced_quota,检查余额并扣减在一条 SQL 中完成。 +// 语义:扣减后 synced_quota 必须 >= threshold,即 synced_quota >= amount + threshold。 // 返回 (当前余额, 是否成功, 错误)。 func AtomicDecreaseSyncedQuota(userId int, amount int, threshold int) (int, bool, error) { result := DB.Model(&User{}). @@ -1021,6 +1022,13 @@ func IncreaseSyncedQuota(userId int, amount int) error { Update("synced_quota", gorm.Expr("synced_quota + ?", amount)).Error } +// AdjustSyncedQuota 原子调整 synced_quota(增量更新,避免并发覆盖) +// delta > 0 表示额外扣减,delta < 0 表示退还。 +func AdjustSyncedQuota(userId int, delta int) error { + return DB.Model(&User{}).Where("id = ?", userId). + Update("synced_quota", gorm.Expr("synced_quota - ?", delta)).Error +} + //func GetRootUserEmail() (email string) { // DB.Model(&User{}).Where("role = ?", common.RoleRootUser).Select("email").Find(&email) // return email diff --git a/service/billing_session_sync_test.go b/service/billing_session_sync_test.go index a6acbfb..615fb4d 100644 --- a/service/billing_session_sync_test.go +++ b/service/billing_session_sync_test.go @@ -26,7 +26,7 @@ func setupBillingSyncDB(t *testing.T) *gorm.DB { common.UsingSQLite = true common.RedisEnabled = false - require.NoError(t, db.AutoMigrate(&model.PendingSyncRecord{})) + require.NoError(t, db.AutoMigrate(&model.PendingSyncRecord{}, &model.User{})) t.Cleanup(func() { model.DB = origDB @@ -45,6 +45,7 @@ func TestNeedsRefundLocked_SyncedUser(t *testing.T) { defer func() { settings.MinBalanceThreshold = origThreshold }() settings.MinBalanceThreshold = 100000 + require.NoError(t, model.DB.Create(&model.User{Id: 100, Username: "test-nr", Source: common.UserSourceSynced, SyncedQuota: 500000}).Error) f := NewSyncedUserFunding(100, 200, "req-nr-001", 500000) require.NoError(t, f.PreConsume(5000)) @@ -75,6 +76,7 @@ func TestNeedsRefundLocked_SyncedUser_FundingSettled(t *testing.T) { defer func() { settings.MinBalanceThreshold = origThreshold }() settings.MinBalanceThreshold = 100000 + require.NoError(t, model.DB.Create(&model.User{Id: 100, Username: "test-nr3", Source: common.UserSourceSynced, SyncedQuota: 500000}).Error) f := NewSyncedUserFunding(100, 200, "req-nr-003", 500000) require.NoError(t, f.PreConsume(5000)) @@ -124,6 +126,7 @@ func TestNeedsRefundLocked_WalletAndSynced(t *testing.T) { defer func() { settings.MinBalanceThreshold = origThreshold }() settings.MinBalanceThreshold = 100000 + require.NoError(t, model.DB.Create(&model.User{Id: 100, Username: "test-compare", Source: common.UserSourceSynced, SyncedQuota: 500000}).Error) syncF := NewSyncedUserFunding(100, 200, "req-compare", 500000) require.NoError(t, syncF.PreConsume(5000)) diff --git a/service/region_sync/sync_manager.go b/service/region_sync/sync_manager.go index a7d3a31..623daeb 100644 --- a/service/region_sync/sync_manager.go +++ b/service/region_sync/sync_manager.go @@ -81,11 +81,10 @@ func (m *SyncManager) RunBatchSync() int { resp, err := m.client.BatchDeduct(&BatchDeductRequest{Records: batchRecords}) if err != nil { common.SysError(fmt.Sprintf("[SyncManager] BatchDeduct failed: %v", err)) + // 网络失败时,所有记录都标记为 failed 并增加 retry_count,防止无限重试 for _, record := range records { - if record.RetryCount+1 >= settings.MaxRetryCount { - if markErr := model.MarkRecordFailed(record.Id, err.Error()); markErr != nil { - common.SysError(fmt.Sprintf("[SyncManager] MarkRecordFailed error: %v", markErr)) - } + if markErr := model.MarkRecordFailed(record.Id, err.Error()); markErr != nil { + common.SysError(fmt.Sprintf("[SyncManager] MarkRecordFailed error: %v", markErr)) } } return 0 @@ -208,6 +207,10 @@ func (m *SyncManager) StartSyncWorkers() { quotaSyncInterval := time.Duration(settings.QuotaSyncIntervalSeconds) * time.Second startWorker(quotaSyncInterval, m.RunQuotaSync, "quota sync completed") startWorker(time.Hour, m.RunCleanup, "cleanup completed") + // 清理超时的 pending 记录(请求崩溃的情况,5分钟超时) + startWorker(5*time.Minute, func() int { + return int(model.ArchiveStalePendingRecords(300)) + }, "stale pending records archived") } // Stop 停止同步管理器 diff --git a/service/region_sync/sync_manager_test.go b/service/region_sync/sync_manager_test.go index affb02f..4c5c55c 100644 --- a/service/region_sync/sync_manager_test.go +++ b/service/region_sync/sync_manager_test.go @@ -64,7 +64,7 @@ func TestSyncManager_RunBatchSync_WithRecords(t *testing.T) { RemoteUserId: 100, RequestId: "req-001", Quota: 100, - Status: model.PendingSyncStatusPending, + Status: model.PendingSyncStatusSettled, CreatedAt: time.Now().Unix(), } record2 := &model.PendingSyncRecord{ @@ -73,7 +73,7 @@ func TestSyncManager_RunBatchSync_WithRecords(t *testing.T) { RemoteUserId: 100, RequestId: "req-002", Quota: 200, - Status: model.PendingSyncStatusPending, + Status: model.PendingSyncStatusSettled, CreatedAt: time.Now().Unix(), } @@ -146,7 +146,7 @@ func TestSyncManager_RunBatchSync_MultipleRecords(t *testing.T) { RemoteUserId: 100, RequestId: fmt.Sprintf("req-batch-%d", i), Quota: i * 100, - Status: model.PendingSyncStatusPending, + Status: model.PendingSyncStatusSettled, CreatedAt: time.Now().Unix(), } require.NoError(t, db.Create(record).Error) @@ -205,11 +205,11 @@ func TestRunBatchSync_WithMockServer(t *testing.T) { require.NoError(t, db.Exec("DELETE FROM pending_sync_records").Error) require.NoError(t, db.Create(&model.PendingSyncRecord{ UserId: 100, RemoteUserId: 200, RequestId: "req-batch-ok-1", - Quota: 100, Status: model.PendingSyncStatusPending, CreatedAt: now, + Quota: 100, Status: model.PendingSyncStatusSettled, CreatedAt: now, }).Error) require.NoError(t, db.Create(&model.PendingSyncRecord{ UserId: 101, RemoteUserId: 201, RequestId: "req-batch-ok-2", - Quota: 200, Status: model.PendingSyncStatusPending, CreatedAt: now + 1, + Quota: 200, Status: model.PendingSyncStatusSettled, CreatedAt: now + 1, }).Error) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -237,11 +237,11 @@ func TestRunBatchSync_WithMockServer(t *testing.T) { require.NoError(t, db.Exec("DELETE FROM pending_sync_records").Error) require.NoError(t, db.Create(&model.PendingSyncRecord{ UserId: 100, RemoteUserId: 200, RequestId: "req-part-1", - Quota: 100, Status: model.PendingSyncStatusPending, CreatedAt: now, + Quota: 100, Status: model.PendingSyncStatusSettled, CreatedAt: now, }).Error) require.NoError(t, db.Create(&model.PendingSyncRecord{ UserId: 101, RemoteUserId: 201, RequestId: "req-part-2", - Quota: 200, Status: model.PendingSyncStatusPending, CreatedAt: now + 1, + Quota: 200, Status: model.PendingSyncStatusSettled, CreatedAt: now + 1, }).Error) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -262,7 +262,7 @@ func TestRunBatchSync_WithMockServer(t *testing.T) { var syncedCount, pendingCount int64 db.Model(&model.PendingSyncRecord{}).Where("status = ?", model.PendingSyncStatusSynced).Count(&syncedCount) - db.Model(&model.PendingSyncRecord{}).Where("status = ?", model.PendingSyncStatusPending).Count(&pendingCount) + db.Model(&model.PendingSyncRecord{}).Where("status = ?", model.PendingSyncStatusSettled).Count(&pendingCount) assert.Equal(t, int64(1), syncedCount) assert.Equal(t, int64(1), pendingCount) }) @@ -271,7 +271,7 @@ func TestRunBatchSync_WithMockServer(t *testing.T) { require.NoError(t, db.Exec("DELETE FROM pending_sync_records").Error) require.NoError(t, db.Create(&model.PendingSyncRecord{ UserId: 100, RemoteUserId: 200, RequestId: "req-srv-err", - Quota: 100, Status: model.PendingSyncStatusPending, CreatedAt: now, + Quota: 100, Status: model.PendingSyncStatusSettled, CreatedAt: now, }).Error) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/service/region_sync/sync_push.go b/service/region_sync/sync_push.go index 20b1732..7b6964a 100644 --- a/service/region_sync/sync_push.go +++ b/service/region_sync/sync_push.go @@ -26,7 +26,7 @@ func PushUserCreateToSlave(user *model.User) { req := &SyncUserRequest{ Username: user.Username, Email: user.Email, - PasswordHash: user.Password, + PasswordHash: user.Password, // 传输的是哈希值,非明文密码 DisplayName: user.DisplayName, Quota: user.Quota, RemoteUserId: user.Id, diff --git a/service/synced_user_funding.go b/service/synced_user_funding.go index 8ea3a54..390ec11 100644 --- a/service/synced_user_funding.go +++ b/service/synced_user_funding.go @@ -52,7 +52,7 @@ func (s *SyncedUserFunding) PreConsume(amount int) error { s.userId, newQuota, amount, settings.MinBalanceThreshold) } - // 创建待同步记录(预估扣费) + // 创建待同步记录(status=pending) err = model.CreatePendingSyncRecord(s.userId, s.remoteUserId, s.requestId, amount, amount) if err != nil { // 回滚:退还已扣减的额度 @@ -74,10 +74,8 @@ func (s *SyncedUserFunding) Settle(delta int) error { // 结算时更新待同步记录的实际扣费额度 if delta > 0 { - // 需要额外扣减 s.consumed += delta } else { - // 需要退还部分额度 s.consumed += delta // delta 是负数 } @@ -86,20 +84,24 @@ func (s *SyncedUserFunding) Settle(delta int) error { common.SysLog(fmt.Sprintf("[RegionSync] Settle: userId=%d, delta=%d, consumed=%d, syncedQuota=%d, requestId=%s", s.userId, delta, s.consumed, s.syncedQuota, s.requestId)) - // 尝试更新已有的 PendingSyncRecord;如果不存在(PreConsume 时 amount=0 跳过),则创建新记录 - updated := model.UpdatePendingSyncRecordQuota(s.requestId, s.consumed) - if updated == 0 { - // PreConsume 跳过时没有创建记录,此处用实际消耗量创建 - if err := model.CreatePendingSyncRecord(s.userId, s.remoteUserId, s.requestId, s.consumed, s.consumed); err != nil { - common.SysError(fmt.Sprintf("[RegionSync] Settle: failed to create pending record, requestId=%s, err=%v", s.requestId, err)) + // 将 pending record 结算为 settled(原子更新 quota + status) + settled, err := model.SettlePendingSyncRecord(s.requestId, s.consumed) + if err != nil { + common.SysError(fmt.Sprintf("[RegionSync] Settle: failed to settle record, requestId=%s, err=%v", s.requestId, err)) + } else if !settled { + // 记录不存在(PreConsume 跳过 amount=0),直接创建 settled 记录 + if err := model.CreateSettledSyncRecord(s.userId, s.remoteUserId, s.requestId, s.consumed); err != nil { + common.SysError(fmt.Sprintf("[RegionSync] Settle: failed to create settled record, requestId=%s, err=%v", s.requestId, err)) } else { - common.SysLog(fmt.Sprintf("[RegionSync] Settle: created pending record (PreConsume skipped), userId=%d, quota=%d, requestId=%s", s.userId, s.consumed, s.requestId)) + common.SysLog(fmt.Sprintf("[RegionSync] Settle: created settled record (PreConsume skipped), userId=%d, quota=%d, requestId=%s", s.userId, s.consumed, s.requestId)) } } - // 更新数据库中用户的 synced_quota 快照 - if err := model.UpdateSyncedQuota(s.userId, s.syncedQuota); err != nil { - common.SysError(fmt.Sprintf("[RegionSync] Settle: failed to update synced_quota, userId=%d, err=%v", s.userId, err)) + // 原子增量更新 DB synced_quota(避免并发请求覆盖彼此的更新) + if delta != 0 { + if err := model.AdjustSyncedQuota(s.userId, delta); err != nil { + common.SysError(fmt.Sprintf("[RegionSync] Settle: failed to adjust synced_quota, userId=%d, delta=%d, err=%v", s.userId, delta, err)) + } } return nil diff --git a/service/synced_user_funding_test.go b/service/synced_user_funding_test.go index 0406ef2..df489fd 100644 --- a/service/synced_user_funding_test.go +++ b/service/synced_user_funding_test.go @@ -27,7 +27,7 @@ func setupSyncedFundingDB(t *testing.T) *gorm.DB { common.UsingSQLite = true common.RedisEnabled = false - require.NoError(t, db.AutoMigrate(&model.PendingSyncRecord{})) + require.NoError(t, db.AutoMigrate(&model.PendingSyncRecord{}, &model.User{})) t.Cleanup(func() { model.DB = origDB @@ -59,6 +59,11 @@ func TestPreConsume_Success(t *testing.T) { settings.MinBalanceThreshold = 100000 defer func() { settings.MinBalanceThreshold = origThreshold }() + // 创建测试用户(AtomicDecreaseSyncedQuota 操作 DB 中的 synced_quota) + require.NoError(t, model.DB.Create(&model.User{ + Id: 100, Username: "test-synced", Source: common.UserSourceSynced, SyncedQuota: 1000000, + }).Error) + f := NewSyncedUserFunding(100, 200, "req-pre-success", 1000000) err := f.PreConsume(50000) require.NoError(t, err) @@ -99,8 +104,12 @@ func TestPreConsume_InsufficientQuota(t *testing.T) { settings.MinBalanceThreshold = 100000 defer func() { settings.MinBalanceThreshold = origThreshold }() - // syncedQuota=150000, amount=50000, threshold=100000 - // 150000 < 50000 + 100000 = 150000 → 不满足(严格小于) + require.NoError(t, model.DB.Create(&model.User{ + Id: 100, Username: "test-insuf", Source: common.UserSourceSynced, SyncedQuota: 149999, + }).Error) + + // syncedQuota=149999, amount=50000, threshold=100000 + // 149999 < 50000 + 100000 = 150000 → 不满足 f := NewSyncedUserFunding(100, 200, "req-pre-insuf", 149999) err := f.PreConsume(50000) assert.Error(t, err) @@ -116,6 +125,10 @@ func TestPreConsume_ExactlyAtThreshold(t *testing.T) { settings.MinBalanceThreshold = 100000 defer func() { settings.MinBalanceThreshold = origThreshold }() + require.NoError(t, model.DB.Create(&model.User{ + Id: 100, Username: "test-exact", Source: common.UserSourceSynced, SyncedQuota: 150000, + }).Error) + // syncedQuota=150000, amount=50000, threshold=100000 // 150000 >= 50000 + 100000 = 150000 → 刚好满足 f := NewSyncedUserFunding(100, 200, "req-pre-exact", 150000) @@ -125,6 +138,8 @@ func TestPreConsume_ExactlyAtThreshold(t *testing.T) { } func TestSettle_PositiveDelta(t *testing.T) { + setupSyncedFundingDB(t) + f := NewSyncedUserFunding(100, 200, "req-settle-pos", 500000) f.consumed = 1000 @@ -135,6 +150,8 @@ func TestSettle_PositiveDelta(t *testing.T) { } func TestSettle_NegativeDelta(t *testing.T) { + setupSyncedFundingDB(t) + f := NewSyncedUserFunding(100, 200, "req-settle-neg", 500000) f.consumed = 1000 @@ -150,13 +167,38 @@ func TestSettle_ZeroDelta(t *testing.T) { err := f.Settle(0) assert.NoError(t, err) - assert.Equal(t, 1000, f.consumed) // 不变 - assert.Equal(t, 500000, f.syncedQuota) // 不变 + assert.Equal(t, 1000, f.consumed) // 不变 + assert.Equal(t, 500000, f.syncedQuota) // 不变 +} + +func TestSettle_TransitionsPendingToSettled(t *testing.T) { + db := setupSyncedFundingDB(t) + + // 先创建 pending 记录 + require.NoError(t, model.CreatePendingSyncRecord(100, 200, "req-settle-trans", 1000, 1000)) + + f := NewSyncedUserFunding(100, 200, "req-settle-trans", 500000) + f.consumed = 1000 + + err := f.Settle(-300) // 实际消费 700,退还 300 + assert.NoError(t, err) + assert.Equal(t, 700, f.consumed) + + // 验证记录状态从 pending 转为 settled,quota 更新为实际消费 + var record model.PendingSyncRecord + require.NoError(t, db.Where("request_id = ?", "req-settle-trans").First(&record).Error) + assert.Equal(t, model.PendingSyncStatusSettled, record.Status) + assert.Equal(t, 700, record.Quota) } func TestRefund_Success(t *testing.T) { db := setupSyncedFundingDB(t) + // 创建测试用户 + require.NoError(t, model.DB.Create(&model.User{ + Id: 100, Username: "test-refund", Source: common.UserSourceSynced, SyncedQuota: 500000, + }).Error) + // 先创建一条 pending 记录 require.NoError(t, model.CreatePendingSyncRecord(100, 200, "req-refund-ok", 5000, 5000))