fix(task): wait up to 72 hours before timing out tasks

Some vendors keep a task running for more than a day: Volcengine Ark waits
48 hours by default and up to 72 when configured, and Vidu off-peak jobs
can take 48 hours. With the 24-hour default the gateway failed and refunded
those tasks while the vendor was still producing, and charging for, the
result. TASK_TIMEOUT_MINUTES now defaults to 4320. A task that is truly
stuck is refunded after 72 hours instead of 24.
This commit is contained in:
CaIon committed 2026-10-24 21:07:04 +08:00
1 parent a404460413
commit 0034b24914
4 files changed
+38 -3

No files matched your search

+1 -1
View File
@@ -59,7 +59,7 @@
# 更新任务启用
# UPDATE_TASK=true
# 异步任务硬超时(分钟),按提交时间计算,超时未完成的任务标记失败并退款;0 表示禁用
# TASK_TIMEOUT_MINUTES=1440
# TASK_TIMEOUT_MINUTES=4320
# 异步任务连续轮询失败阈值(上游 429/5xx/401/403、网络错误、无法识别的响应),达到后任务标记失败并退款;正常轮询成功一次即归零
# TASK_POLL_MAX_FAILURES=20
+1 -1
View File
@@ -206,7 +206,7 @@ func initConstantEnv() {
// 任务轮询时查询的最大数量
constant.TaskQueryLimit = GetEnvOrDefault("TASK_QUERY_LIMIT", 1000)
// 异步任务超时时间(分钟),超过此时间未完成的任务将被标记为失败并退款。0 表示禁用。
constant.TaskTimeoutMinutes = GetEnvOrDefault("TASK_TIMEOUT_MINUTES", 1440)
constant.TaskTimeoutMinutes = GetEnvOrDefault("TASK_TIMEOUT_MINUTES", 4320)
// Consecutive unrecognized/transient poll failures before the task is failed and refunded.
constant.TaskPollMaxFailures = GetEnvOrDefault("TASK_POLL_MAX_FAILURES", 20)
// 声明式任务协议桥只观察数据库;这些值控制一次客户端观察连接,
+2 -1
View File
@@ -776,7 +776,8 @@ func unrecognizedPollDetail(reason string, body []byte) string {
func recordPollFailure(ctx context.Context, adaptor TaskPollingAdaptor, task *model.Task, fromStatus model.TaskStatus, class string, statusCode int, detail string) error {
task.PrivateData.PollFailures++
// TASK_POLL_MAX_FAILURES <= 0 disables the consecutive-failure cutoff, matching
// TASK_TIMEOUT_MINUTES semantics; the 24h sweep remains the only backstop.
// TASK_TIMEOUT_MINUTES semantics; the 72h sweep (TASK_TIMEOUT_MINUTES default
// 4320) remains the only backstop.
giveUp := constant.TaskPollMaxFailures > 0 && task.PrivateData.PollFailures >= constant.TaskPollMaxFailures
if giveUp || class == pollClassUnrecognized || class == pollClassHookError {
// The redacted body is intentionally not persisted to Task.Data on these
+34
View File
@@ -775,6 +775,40 @@ func TestSweepTimedOutTasksHonorsRefundRolloutBoundary(t *testing.T) {
assert.Equal(t, int64(1), countLogs(t))
}
// TASK_TIMEOUT_MINUTES defaults to 72 hours, the longest a vendor keeps a
// task running (Ark can be configured to 72h, Vidu off-peak runs 48h).
func TestSweepTimedOutTasksAtTheDefaultDeadline(t *testing.T) {
truncate(t)
const userID, initialQuota, taskQuota = 404, 10_000, 1_500
seedUser(t, userID, initialQuota)
running := makeTask(userID, 0, taskQuota, 0, BillingSourceWallet, 0)
running.TaskID = "running_for_60_hours"
running.Progress = "50%"
running.SubmitTime = time.Now().Add(-60 * time.Hour).Unix()
require.NoError(t, model.DB.Create(running).Error)
stuck := makeTask(userID, 0, taskQuota, 0, BillingSourceWallet, 0)
stuck.TaskID = "running_for_73_hours"
stuck.Progress = "50%"
stuck.SubmitTime = time.Now().Add(-73 * time.Hour).Unix()
require.NoError(t, model.DB.Create(stuck).Error)
previousTimeout := constant.TaskTimeoutMinutes
constant.TaskTimeoutMinutes = 4320
t.Cleanup(func() { constant.TaskTimeoutMinutes = previousTimeout })
sweepTimedOutTasks(context.Background())
var reloadedRunning, reloadedStuck model.Task
require.NoError(t, model.DB.First(&reloadedRunning, running.ID).Error)
require.NoError(t, model.DB.First(&reloadedStuck, stuck.ID).Error)
assert.NotEqualValues(t, model.TaskStatusFailure, reloadedRunning.Status)
assert.Equal(t, taskQuota, reloadedRunning.Quota)
assert.EqualValues(t, model.TaskStatusFailure, reloadedStuck.Status)
assert.Equal(t, "Task timed out", reloadedStuck.FailReason)
assert.Equal(t, initialQuota+taskQuota, getUserQuota(t, userID))
}
type scriptedPollingAdaptor struct {
statusCode int
body []byte