fix(worker): 隔离任务创建与队列恢复窗口
高媒体并发下,任务创建后的请求恢复和候选准入可能持续数十秒;通用 River 恢复扫描会在准入记录落库前误把新任务当成崩溃遗留任务,提前创建 Job 并占满 Worker 执行槽。\n\n为无活动 River Job 的通用兜底恢复增加 5 分钟新任务保护窗。正常提交、准入调度以及已有 River Job 的 Worker 强杀恢复保持原路径;新增 PostgreSQL 集成回归,验证新任务不被抢占且超过保护窗后仍可恢复。\n\n验证:Go 全量测试、go vet、runner/store race、gofmt、迁移安全检查通过。
This commit is contained in:
@@ -471,6 +471,36 @@ WHERE id = $1::uuid`, queuedAtomicTask.ID).Scan(&riverJobID, &queuedLeases); err
|
|||||||
t.Fatal("generic River recovery claimed a task owned by the async admission dispatcher")
|
t.Fatal("generic River recovery claimed a task owned by the async admission dispatcher")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
recoveryGraceTask := createTask(true)
|
||||||
|
recoverableTasks, err = first.ListRecoverableAsyncTasks(ctx, 1000)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("list fresh recoverable tasks: %v", err)
|
||||||
|
}
|
||||||
|
for _, item := range recoverableTasks {
|
||||||
|
if item.ID == recoveryGraceTask.ID {
|
||||||
|
t.Fatal("generic River recovery claimed a task still preparing admission")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if _, err := first.pool.Exec(ctx, `
|
||||||
|
UPDATE gateway_tasks
|
||||||
|
SET created_at = now() - interval '6 minutes'
|
||||||
|
WHERE id = $1::uuid`, recoveryGraceTask.ID); err != nil {
|
||||||
|
t.Fatalf("age task beyond River recovery grace period: %v", err)
|
||||||
|
}
|
||||||
|
recoverableTasks, err = first.ListRecoverableAsyncTasks(ctx, 1000)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("list aged recoverable tasks: %v", err)
|
||||||
|
}
|
||||||
|
recoveryGraceFound := false
|
||||||
|
for _, item := range recoverableTasks {
|
||||||
|
if item.ID == recoveryGraceTask.ID {
|
||||||
|
recoveryGraceFound = true
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if !recoveryGraceFound {
|
||||||
|
t.Fatal("generic River recovery did not claim a task beyond the recovery grace period")
|
||||||
|
}
|
||||||
result, err = second.TryTaskAdmission(ctx, inputFor(queuedAtomicTask, 100, ""))
|
result, err = second.TryTaskAdmission(ctx, inputFor(queuedAtomicTask, 100, ""))
|
||||||
if err != nil || !result.Admitted || len(result.Leases) != 1 {
|
if err != nil || !result.Admitted || len(result.Leases) != 1 {
|
||||||
t.Fatalf("worker-time queued admission result=%+v err=%v", result, err)
|
t.Fatalf("worker-time queued admission result=%+v err=%v", result, err)
|
||||||
|
|||||||
@@ -1043,11 +1043,17 @@ func (s *Store) ListRecoverableAsyncTasks(ctx context.Context, limit int) ([]Asy
|
|||||||
if limit <= 0 {
|
if limit <= 0 {
|
||||||
limit = 500
|
limit = 500
|
||||||
}
|
}
|
||||||
|
// API handlers create the durable task before they finish media restoration
|
||||||
|
// and register distributed admission. Give that normal submission path
|
||||||
|
// ownership of fresh tasks so the generic crash recovery scan cannot create
|
||||||
|
// a River job inside the create-to-admission window.
|
||||||
|
const minimumRecoveryAge = 5 * time.Minute
|
||||||
rows, err := s.pool.Query(ctx, `
|
rows, err := s.pool.Query(ctx, `
|
||||||
SELECT task.id::text, task.priority, task.next_run_at
|
SELECT task.id::text, task.priority, task.next_run_at
|
||||||
FROM gateway_tasks task
|
FROM gateway_tasks task
|
||||||
LEFT JOIN river_job job ON job.id = task.river_job_id
|
LEFT JOIN river_job job ON job.id = task.river_job_id
|
||||||
WHERE task.async_mode = true
|
WHERE task.async_mode = true
|
||||||
|
AND task.created_at <= now() - $2::interval
|
||||||
AND (
|
AND (
|
||||||
task.status = 'queued'
|
task.status = 'queued'
|
||||||
OR (
|
OR (
|
||||||
@@ -1067,7 +1073,7 @@ WHERE task.async_mode = true
|
|||||||
AND admission.status = 'waiting'
|
AND admission.status = 'waiting'
|
||||||
)
|
)
|
||||||
ORDER BY task.priority ASC, task.created_at ASC
|
ORDER BY task.priority ASC, task.created_at ASC
|
||||||
LIMIT $1`, limit)
|
LIMIT $1`, limit, minimumRecoveryAge.String())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user