Skip to content

phsl / new-api

Go to a project
Commit bc14c18f by CaIon
Options

refactor: update task refund logic and remove legacy handling

parent 2d23cdf2
Showing with 62 additions and 308 deletions
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 }
......
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
......
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())
}
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()))
......
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)
......
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)
task := makeTask(userID, 0, taskQuota, 0, BillingSourceWallet, 0)
task.TaskID = "historical_failed_already_refunded"
task.Status = model.TaskStatusFailure
task.Progress = "100%"
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)
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)
previousFactory := GetTaskAdaptorFunc
GetTaskAdaptorFunc = func(constant.TaskPlatform) TaskPollingAdaptor {
return &taskPollingFetchAdaptor{}
}
t.Cleanup(func() { GetTaskAdaptorFunc = previousFactory })
sweepUnrefundedFailedTasks(context.Background())
sweepUnrefundedFailedTasks(context.Background())
summary := RunTaskPollingOnce(context.Background(), nil)
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))
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))
}
func TestSweepUnrefundedFailedTasksRestoresMarkerAfterFundingFailure(t *testing.T) {
func TestSweepTimedOutTasksHonorsRefundRolloutBoundary(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.Status = model.TaskStatusFailure
task.Progress = "100%"
task.SubmitTime = model.TaskRefundLegacyCutoff
task.UpdatedAt = time.Now().Add(-time.Minute).Unix()
require.NoError(t, model.DB.Create(task).Error)
const (
userID = 403
initialQuota = 10_000
legacyTaskQuota = 1_800
modernTaskQuota = 1_200
)
seedUser(t, userID, initialQuota)
sweepUnrefundedFailedTasks(context.Background())
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)
var afterFailedRefund model.Task
require.NoError(t, model.DB.First(&afterFailedRefund, task.ID).Error)
assert.Equal(t, taskQuota, afterFailedRefund.Quota)
assert.Equal(t, int64(0), countLogs(t))
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)
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)
previousTimeout := constant.TaskTimeoutMinutes
constant.TaskTimeoutMinutes = 1
t.Cleanup(func() { constant.TaskTimeoutMinutes = previousTimeout })
sweepUnrefundedFailedTasks(context.Background())
sweepTimedOutTasks(context.Background())
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))
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))
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment