From bc14c18f6024e79cba1c08d02cd007796e12d668 Mon Sep 17 00:00:00 2001 From: CaIon Date: Sun, 26 Jul 2026 20:47:52 +0800 Subject: [PATCH] refactor: update task refund logic and remove legacy handling --- controller/system_task_handlers.go | 2 +- model/task.go | 104 +-------------------------- model/task_cas_test.go | 105 --------------------------- service/task_billing.go | 6 +- service/task_polling.go | 46 +----------- service/task_polling_test.go | 111 ++++++++++++++--------------- 6 files changed, 64 insertions(+), 310 deletions(-) diff --git a/controller/system_task_handlers.go b/controller/system_task_handlers.go index 2168b451..c31059d1 100644 --- a/controller/system_task_handlers.go +++ b/controller/system_task_handlers.go @@ -140,7 +140,7 @@ type asyncTaskPollHandler struct{} func (asyncTaskPollHandler) Type() string { return model.SystemTaskTypeAsyncTaskPoll } func (asyncTaskPollHandler) Enabled() bool { - return constant.UpdateTask && model.HasTaskPollingWork() + return constant.UpdateTask && model.HasUnfinishedSyncTasks() } func (asyncTaskPollHandler) Interval() time.Duration { return 15 * time.Second } diff --git a/model/task.go b/model/task.go index ecaf70f3..b50d45b9 100644 --- a/model/task.go +++ b/model/task.go @@ -41,9 +41,9 @@ const ( TaskStatusUnknown = "UNKNOWN" ) -// TaskRefundLegacyCutoff separates legacy timeout tasks that intentionally -// do not receive automatic refunds from tasks covered by reconciliation. -const TaskRefundLegacyCutoff int64 = 1740182400 // 2025-02-22 00:00:00 UTC +// TaskRefundLegacyCutoff separates tasks created before timeout refunds were +// introduced. Those legacy tasks are failed without an automatic refund. +const TaskRefundLegacyCutoff int64 = 1771718400 // 2026-02-22 00:00:00 UTC type Task struct { ID int64 `json:"id" gorm:"primary_key;AUTO_INCREMENT"` @@ -308,28 +308,6 @@ func GetTimedOutUnfinishedTasks(cutoffUnix int64, limit int) []*Task { return tasks } -// GetUnrefundedFailedTasks returns failed tasks whose non-zero quota marks a -// pending refund. Legacy timeout tasks are excluded before LIMIT is applied so -// they cannot starve refundable tasks from the reconciliation sweep. -func GetUnrefundedFailedTasks(updatedBefore int64, limit int) []*Task { - if limit <= 0 { - return nil - } - - var tasks []*Task - err := DB.Where("status = ?", TaskStatusFailure). - Where("quota != ?", 0). - Where("updated_at <= ?", updatedBefore). - Where("(submit_time <= ? OR submit_time >= ?)", 0, TaskRefundLegacyCutoff). - Order("id"). - Limit(limit). - Find(&tasks).Error - if err != nil { - return nil - } - return tasks -} - func GetAllUnFinishSyncTasks(limit int) []*Task { var tasks []*Task var err error @@ -356,38 +334,6 @@ func HasUnfinishedSyncTasks() bool { return err == nil && id != 0 } -// HasTaskPollingWork reports whether polling has either an unfinished task or -// a failed task with a pending, non-legacy refund. The latter keeps the system -// task scheduler active when reconciliation is the only work left. -func HasTaskPollingWork() bool { - if HasUnfinishedSyncTasks() { - return true - } - - var id int64 - err := DB.Model(&Task{}). - Where("status = ?", TaskStatusFailure). - Where("quota != ?", 0). - Where("(submit_time <= ? OR submit_time >= ?)", 0, TaskRefundLegacyCutoff). - Limit(1). - Pluck("id", &id).Error - return err == nil && id != 0 -} - -func GetByOnlyTaskId(taskId string) (*Task, bool, error) { - if taskId == "" { - return nil, false, nil - } - var task *Task - var err error - err = DB.Where("task_id = ?", taskId).First(&task).Error - exist, err := RecordExist(err) - if err != nil { - return nil, false, err - } - return task, exist, err -} - func GetByTaskId(userId int, taskId string) (*Task, bool, error) { if taskId == "" { return nil, false, nil @@ -465,39 +411,6 @@ func (t *Task) UpdateQuota() error { return DB.Model(t).Update("quota", t.Quota).Error } -// ClaimQuotaForRefund atomically clears an expected non-zero quota. A true -// result grants the caller ownership of the corresponding refund attempt. -func ClaimQuotaForRefund(id int64, expectedQuota int) (bool, error) { - if expectedQuota == 0 { - return false, nil - } - - result := DB.Model(&Task{}). - Where("id = ? AND quota = ?", id, expectedQuota). - Update("quota", 0) - if result.Error != nil { - return false, result.Error - } - return result.RowsAffected > 0, nil -} - -// RestoreQuotaAfterFailedRefund restores a claimed quota marker only while it -// is still zero. It is used when the observable funding adjustment fails, so a -// later reconciliation pass can retry without overwriting another writer. -func RestoreQuotaAfterFailedRefund(id int64, quota int) (bool, error) { - if quota == 0 { - return false, nil - } - - result := DB.Model(&Task{}). - Where("id = ? AND quota = ?", id, 0). - Update("quota", quota) - if result.Error != nil { - return false, result.Error - } - return result.RowsAffected > 0, nil -} - // UpdateWithStatus performs a conditional UPDATE guarded by fromStatus (CAS). // Returns (true, nil) if this caller won the update, (false, nil) if // another process already moved the task out of fromStatus. MySQL commonly @@ -515,17 +428,6 @@ func (t *Task) UpdateWithStatus(fromStatus TaskStatus) (bool, error) { return result.RowsAffected > 0, nil } -// TaskBulkUpdate performs an unconditional bulk UPDATE by upstream task_id strings. -// Same caveats as TaskBulkUpdateByID — no CAS guard. -func TaskBulkUpdate(taskIds []string, params map[string]any) error { - if len(taskIds) == 0 { - return nil - } - return DB.Model(&Task{}). - Where("task_id in (?)", taskIds). - Updates(params).Error -} - // TaskBulkUpdateByID performs an unconditional bulk UPDATE by primary key IDs. // WARNING: This function has NO CAS (Compare-And-Swap) guard — it will overwrite // any concurrent status changes. DO NOT use in billing/quota lifecycle flows diff --git a/model/task_cas_test.go b/model/task_cas_test.go index d99a8d89..a53804d3 100644 --- a/model/task_cas_test.go +++ b/model/task_cas_test.go @@ -256,108 +256,3 @@ func TestUpdateWithStatus_ConcurrentWinner(t *testing.T) { } assert.Equal(t, 1, winCount, "exactly one goroutine should win the CAS") } - -func TestClaimQuotaForRefund_OnlyOneClaimSucceeds(t *testing.T) { - truncateTables(t) - - task := &Task{ - TaskID: "task_refund_claim", - Status: TaskStatusFailure, - Quota: 1000, - Data: json.RawMessage(`{}`), - } - insertTask(t, task) - - claimed, err := ClaimQuotaForRefund(task.ID, task.Quota) - require.NoError(t, err) - assert.True(t, claimed) - - claimed, err = ClaimQuotaForRefund(task.ID, task.Quota) - require.NoError(t, err) - assert.False(t, claimed) - - var reloaded Task - require.NoError(t, DB.First(&reloaded, task.ID).Error) - assert.Zero(t, reloaded.Quota) -} - -func TestGetUnrefundedFailedTasks_FiltersAndLimits(t *testing.T) { - truncateTables(t) - - tasks := []*Task{ - {TaskID: "failed_refundable_1", Status: TaskStatusFailure, Quota: 100, SubmitTime: TaskRefundLegacyCutoff, Data: json.RawMessage(`{}`)}, - {TaskID: "failed_refundable_2", Status: TaskStatusFailure, Quota: 200, SubmitTime: TaskRefundLegacyCutoff + 1, Data: json.RawMessage(`{}`)}, - {TaskID: "legacy_failed", Status: TaskStatusFailure, Quota: 400, SubmitTime: TaskRefundLegacyCutoff - 1, Data: json.RawMessage(`{}`)}, - {TaskID: "failed_without_quota", Status: TaskStatusFailure, Quota: 0, Data: json.RawMessage(`{}`)}, - {TaskID: "successful_with_quota", Status: TaskStatusSuccess, Quota: 300, Data: json.RawMessage(`{}`)}, - } - for _, task := range tasks { - insertTask(t, task) - } - - updatedBefore := time.Now().Unix() + 1 - found := GetUnrefundedFailedTasks(updatedBefore, 1) - require.Len(t, found, 1) - assert.Equal(t, tasks[0].ID, found[0].ID) - - found = GetUnrefundedFailedTasks(updatedBefore, 10) - require.Len(t, found, 2) - assert.Equal(t, []int64{tasks[0].ID, tasks[1].ID}, []int64{found[0].ID, found[1].ID}) - - assert.Empty(t, GetUnrefundedFailedTasks(updatedBefore, 0)) -} - -func TestRestoreQuotaAfterFailedRefund_OnlyRestoresClaimedMarker(t *testing.T) { - truncateTables(t) - - task := &Task{ - TaskID: "task_refund_restore", - Status: TaskStatusFailure, - Quota: 750, - Data: json.RawMessage(`{}`), - } - insertTask(t, task) - - claimed, err := ClaimQuotaForRefund(task.ID, task.Quota) - require.NoError(t, err) - require.True(t, claimed) - - restored, err := RestoreQuotaAfterFailedRefund(task.ID, task.Quota) - require.NoError(t, err) - assert.True(t, restored) - - restored, err = RestoreQuotaAfterFailedRefund(task.ID, task.Quota) - require.NoError(t, err) - assert.False(t, restored) - - var reloaded Task - require.NoError(t, DB.First(&reloaded, task.ID).Error) - assert.Equal(t, task.Quota, reloaded.Quota) -} - -func TestHasTaskPollingWork_IncludesOnlyRefundableFailedTasks(t *testing.T) { - truncateTables(t) - assert.False(t, HasTaskPollingWork()) - - legacy := &Task{ - TaskID: "legacy_failed_work", - Status: TaskStatusFailure, - Progress: "100%", - Quota: 500, - SubmitTime: TaskRefundLegacyCutoff - 1, - Data: json.RawMessage(`{}`), - } - insertTask(t, legacy) - assert.False(t, HasTaskPollingWork()) - - refundable := &Task{ - TaskID: "refundable_failed_work", - Status: TaskStatusFailure, - Progress: "100%", - Quota: 500, - SubmitTime: TaskRefundLegacyCutoff, - Data: json.RawMessage(`{}`), - } - insertTask(t, refundable) - assert.True(t, HasTaskPollingWork()) -} diff --git a/service/task_billing.go b/service/task_billing.go index 51fb77c3..b677cb62 100644 --- a/service/task_billing.go +++ b/service/task_billing.go @@ -162,7 +162,7 @@ func taskModelName(task *model.Task) string { // RefundTaskQuota 统一的任务失败退款逻辑。 // 当异步任务失败时，将预扣的 quota 退还给用户（支持钱包和订阅），并退还令牌额度。 -// 返回资金来源是否已成功退还；失败时保留 quota 作为后续对账标记。 +// 返回资金来源是否已成功退还；失败时保留 quota，供显式重试或人工对账。 func RefundTaskQuota(ctx context.Context, task *model.Task, reason string) bool { quota := task.Quota if quota == 0 { @@ -194,8 +194,8 @@ func RefundTaskQuota(ctx context.Context, task *model.Task, reason string) bool Other: other, }) - // 4. 资金退款完成后再清除持久化标记；失败时保留非零 quota， - // 由后续对账重试。回写失败必须显式告警，避免漏掉潜在的重复退款风险。 + // 4. 资金退款完成后再清除持久化标记。 + // 回写失败必须显式告警，避免漏掉潜在的重复退款风险。 task.Quota = 0 if err := task.UpdateQuota(); err != nil { logger.LogError(ctx, fmt.Sprintf("退款成功但清除 task quota 失败 task %s: %s", task.TaskID, err.Error())) diff --git a/service/task_polling.go b/service/task_polling.go index 0ee3be24..c6ffeec8 100644 --- a/service/task_polling.go +++ b/service/task_polling.go @@ -37,11 +37,6 @@ type TaskPollingAdaptor interface { // 打破 service -> relay -> relay/channel -> service 的循环依赖。 var GetTaskAdaptorFunc func(platform constant.TaskPlatform) TaskPollingAdaptor -const ( - refundReconciliationLimit = 100 - refundReconciliationGracePeriod = 30 * time.Second -) - // sweepTimedOutTasks 在主轮询之前独立清理超时任务。 // 每次最多处理 100 条，剩余的下个周期继续处理。 // 使用 per-task CAS (UpdateWithStatus) 防止覆盖被正常轮询已推进的任务。 @@ -69,7 +64,8 @@ func sweepTimedOutTasks(ctx context.Context) { task.FinishTime = now if isLegacy { task.FailReason = legacyReason - // 旧系统任务明确不退款，随终态 CAS 一并清掉 quota，避免被后续对账误判。 + // 旧系统任务明确不退款，随终态 CAS 一并清掉 quota， + // 避免留下可再次退款的计费状态。 task.Quota = 0 } else { task.FailReason = reason @@ -95,43 +91,6 @@ func sweepTimedOutTasks(ctx context.Context) { } } -// sweepUnrefundedFailedTasks 重试已落 FAILURE 终态但仍保留 quota 的欠退款任务。 -// 先等待一个短暂宽限期，让终态 CAS 的胜出者完成主路径即时退款，避免正常 -// 轮询与对账同时处理刚失败的任务。 -func sweepUnrefundedFailedTasks(ctx context.Context) { - updatedBefore := time.Now().Add(-refundReconciliationGracePeriod).Unix() - tasks := model.GetUnrefundedFailedTasks(updatedBefore, refundReconciliationLimit) - for _, task := range tasks { - if ctx.Err() != nil { - return - } - - quota := task.Quota - claimed, err := model.ClaimQuotaForRefund(task.ID, quota) - if err != nil { - logger.LogError(ctx, fmt.Sprintf("sweepUnrefundedFailedTasks claim error for task %s: %v", task.TaskID, err)) - continue - } - if !claimed { - logger.LogDebug(ctx, "sweepUnrefundedFailedTasks: task %s claim lost, skip refund", task.TaskID) - continue - } - - // 对账先清 marker 再退款，确保并发 sweep 只有一个实际退款者。若进程在 - // claim 后、退款前崩溃，会偏向漏退而不是双退，需由人工账务对账兜底。 - if RefundTaskQuota(ctx, task, task.FailReason) { - continue - } - - restored, restoreErr := model.RestoreQuotaAfterFailedRefund(task.ID, quota) - if restoreErr != nil { - logger.LogError(ctx, fmt.Sprintf("sweepUnrefundedFailedTasks restore quota error for task %s: %v", task.TaskID, restoreErr)) - } else if !restored { - logger.LogError(ctx, fmt.Sprintf("sweepUnrefundedFailedTasks could not restore quota marker for task %s", task.TaskID)) - } - } -} - // TaskPollSummary is the result recorded on an async_task_poll system task row, // summarizing one polling pass. type TaskPollSummary struct { @@ -156,7 +115,6 @@ func RunTaskPollingOnce(ctx context.Context, report func(processed, total int)) common.SysLog("任务进度轮询开始") sweepTimedOutTasks(ctx) - sweepUnrefundedFailedTasks(ctx) allTasks := model.GetAllUnFinishSyncTasks(constant.TaskQueryLimit) summary.UnfinishedTasks = len(allTasks) platformTask := make(map[constant.TaskPlatform][]*model.Task) diff --git a/service/task_polling_test.go b/service/task_polling_test.go index dd70bf81..228eb40f 100644 --- a/service/task_polling_test.go +++ b/service/task_polling_test.go @@ -395,7 +395,7 @@ func TestUpdateSunoTasksStalePollsRefundExactlyOnce(t *testing.T) { task.Platform = constant.TaskPlatformSuno task.Status = model.TaskStatusInProgress task.Progress = "50%" - task.SubmitTime = model.TaskRefundLegacyCutoff + task.SubmitTime = time.Now().Unix() task.PrivateData.UpstreamTaskID = upstreamTaskID require.NoError(t, model.DB.Create(task).Error) @@ -425,74 +425,73 @@ func TestUpdateSunoTasksStalePollsRefundExactlyOnce(t *testing.T) { assert.Equal(t, int64(1), countLogs(t)) } -func TestSweepUnrefundedFailedTasksRefundsModernTaskAndSkipsLegacy(t *testing.T) { +func TestRunTaskPollingOnceDoesNotRefundHistoricalFailedTask(t *testing.T) { truncate(t) - const userID = 402 - const initialQuota, modernTaskQuota, legacyTaskQuota = 10_000, 1_200, 1_800 + const userID, initialQuota, taskQuota = 402, 10_000, 1_200 seedUser(t, userID, initialQuota) - modernTask := makeTask(userID, 0, modernTaskQuota, 0, BillingSourceWallet, 0) - modernTask.TaskID = "modern_failed_pending_refund" - modernTask.Status = model.TaskStatusFailure - modernTask.Progress = "100%" - modernTask.SubmitTime = model.TaskRefundLegacyCutoff - modernTask.UpdatedAt = time.Now().Add(-time.Minute).Unix() - require.NoError(t, model.DB.Create(modernTask).Error) - - legacyTask := makeTask(userID, 0, legacyTaskQuota, 0, BillingSourceWallet, 0) - legacyTask.TaskID = "legacy_failed_without_refund" - legacyTask.Status = model.TaskStatusFailure - legacyTask.Progress = "100%" - legacyTask.SubmitTime = model.TaskRefundLegacyCutoff - 1 - legacyTask.UpdatedAt = time.Now().Add(-time.Minute).Unix() - require.NoError(t, model.DB.Create(legacyTask).Error) - - sweepUnrefundedFailedTasks(context.Background()) - sweepUnrefundedFailedTasks(context.Background()) - - var reloadedModern model.Task - var reloadedLegacy model.Task - require.NoError(t, model.DB.First(&reloadedModern, modernTask.ID).Error) - require.NoError(t, model.DB.First(&reloadedLegacy, legacyTask.ID).Error) - assert.Zero(t, reloadedModern.Quota) - assert.Equal(t, legacyTaskQuota, reloadedLegacy.Quota) - assert.Equal(t, initialQuota+modernTaskQuota, getUserQuota(t, userID)) - assert.Equal(t, int64(1), countLogs(t)) -} - -func TestSweepUnrefundedFailedTasksRestoresMarkerAfterFundingFailure(t *testing.T) { - truncate(t) - - const userID, subscriptionID, taskQuota = 404, 404, 900 - const subscriptionUsed int64 = 5_000 - seedUser(t, userID, 0) - - task := makeTask(userID, 0, taskQuota, 0, BillingSourceSubscription, subscriptionID) - task.TaskID = "subscription_failed_pending_refund" + task := makeTask(userID, 0, taskQuota, 0, BillingSourceWallet, 0) + task.TaskID = "historical_failed_already_refunded" task.Status = model.TaskStatusFailure task.Progress = "100%" - task.SubmitTime = model.TaskRefundLegacyCutoff + task.SubmitTime = time.Now().Add(-90 * 24 * time.Hour).Unix() task.UpdatedAt = time.Now().Add(-time.Minute).Unix() require.NoError(t, model.DB.Create(task).Error) - sweepUnrefundedFailedTasks(context.Background()) + previousFactory := GetTaskAdaptorFunc + GetTaskAdaptorFunc = func(constant.TaskPlatform) TaskPollingAdaptor { + return &taskPollingFetchAdaptor{} + } + t.Cleanup(func() { GetTaskAdaptorFunc = previousFactory }) - var afterFailedRefund model.Task - require.NoError(t, model.DB.First(&afterFailedRefund, task.ID).Error) - assert.Equal(t, taskQuota, afterFailedRefund.Quota) + summary := RunTaskPollingOnce(context.Background(), nil) + + assert.Zero(t, summary.UnfinishedTasks) + assert.Equal(t, initialQuota, getUserQuota(t, userID)) + assert.Equal(t, taskQuota, getTaskQuota(t, task.ID)) assert.Equal(t, int64(0), countLogs(t)) +} - seedSubscription(t, subscriptionID, userID, 10_000, subscriptionUsed) - require.NoError(t, model.DB.Model(&model.Task{}). - Where("id = ?", task.ID). - UpdateColumn("updated_at", time.Now().Add(-time.Minute).Unix()).Error) +func TestSweepTimedOutTasksHonorsRefundRolloutBoundary(t *testing.T) { + truncate(t) - sweepUnrefundedFailedTasks(context.Background()) + const ( + userID = 403 + initialQuota = 10_000 + legacyTaskQuota = 1_800 + modernTaskQuota = 1_200 + ) + seedUser(t, userID, initialQuota) - var afterSuccessfulRetry model.Task - require.NoError(t, model.DB.First(&afterSuccessfulRetry, task.ID).Error) - assert.Zero(t, afterSuccessfulRetry.Quota) - assert.Equal(t, subscriptionUsed-int64(taskQuota), getSubscriptionUsed(t, subscriptionID)) + legacyTask := makeTask(userID, 0, legacyTaskQuota, 0, BillingSourceWallet, 0) + legacyTask.TaskID = "legacy_timeout_without_refund" + legacyTask.Progress = "50%" + legacyTask.SubmitTime = 1771718399 // 2026-02-21 23:59:59 UTC + require.NoError(t, model.DB.Create(legacyTask).Error) + + modernTask := makeTask(userID, 0, modernTaskQuota, 0, BillingSourceWallet, 0) + modernTask.TaskID = "modern_timeout_with_refund" + modernTask.Progress = "50%" + modernTask.SubmitTime = 1771718400 // 2026-02-22 00:00:00 UTC + require.NoError(t, model.DB.Create(modernTask).Error) + + previousTimeout := constant.TaskTimeoutMinutes + constant.TaskTimeoutMinutes = 1 + t.Cleanup(func() { constant.TaskTimeoutMinutes = previousTimeout }) + + sweepTimedOutTasks(context.Background()) + + var reloadedLegacy model.Task + var reloadedModern model.Task + require.NoError(t, model.DB.First(&reloadedLegacy, legacyTask.ID).Error) + require.NoError(t, model.DB.First(&reloadedModern, modernTask.ID).Error) + assert.EqualValues(t, model.TaskStatusFailure, reloadedLegacy.Status) + assert.EqualValues(t, model.TaskStatusFailure, reloadedModern.Status) + assert.Zero(t, reloadedLegacy.Quota) + assert.Zero(t, reloadedModern.Quota) + assert.Contains(t, reloadedLegacy.FailReason, "旧系统遗留任务") + assert.Contains(t, reloadedModern.FailReason, "任务超时") + assert.Equal(t, initialQuota+modernTaskQuota, getUserQuota(t, userID)) assert.Equal(t, int64(1), countLogs(t)) }