From 04d8b85eec0e923f398be798be7dfce22e65384a Mon Sep 17 00:00:00 2001 From: AetherNo2332 <130164460+AetherNo2332@users.noreply.github.com> Date: Mon, 21 Sep 2026 07:42:07 +0800 Subject: [PATCH] =?UTF-8?q?fix(app):=20=E4=BB=BB=E5=8A=A1=E7=A7=9F?= =?UTF-8?q?=E7=BA=A6=20-=20=E7=BB=AD=E7=A7=9F=E4=B8=8D=E5=86=8D=E8=A2=AB?= =?UTF-8?q?=E8=87=AA=E5=B7=B1=E7=9A=84=E6=89=A7=E8=A1=8C=E6=97=B6=E9=99=90?= =?UTF-8?q?=E6=89=93=E6=96=AD=20(#567)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Original-Commit: 2b796c403f0c99f3b607b53650b42078964ea753 fix(app): 任务租约 - 续租不再被自己的执行时限打断 续租 context 原来派生自任务执行时限(context.WithTimeout(ctx, 5s)):执行时限一到, 续租请求立刻被取消,失败后被误判成"租约失效、等待其他 worker 恢复",任务停在 running, 租约过期后又被别的 worker 重跑(实测一次上游调用被重跑成三次)。 - taskLeaseRenewContext:用 context.WithoutCancel(parent) + 5s,只继承值、不继承取消与时限; - 收尾处若我方执行时限已到点,不再直接返回租约错误,而是继续走失败收尾(真丢失时行为不变)。 新增用例在旧写法下失败、修复后通过。 --- backend/internal/app/task_worker.go | 21 ++++++++++-- backend/internal/app/task_worker_test.go | 41 ++++++++++++++++++++++++ 2 files changed, 59 insertions(+), 3 deletions(-) create mode 100644 backend/internal/app/task_worker_test.go diff --git a/backend/internal/app/task_worker.go b/backend/internal/app/task_worker.go index debe5503f..ea51d3fae 100644 --- a/backend/internal/app/task_worker.go +++ b/backend/internal/app/task_worker.go @@ -148,7 +148,7 @@ func (w *taskWorkerCoordinator) processClaimedTask(task *model.Task, globalSlot for { select { case <-ticker.C: - renewCtx, cancelRenew := context.WithTimeout(ctx, 5*time.Second) + renewCtx, cancelRenew := taskLeaseRenewContext(ctx) var err error if globalSlot != nil { err = globalSlot.Renew(renewCtx) @@ -227,10 +227,16 @@ func (w *taskWorkerCoordinator) processClaimedTask(task *model.Task, globalSlot if code, _ := ChannelSlotFailureDetails(err); code != "" { channelSlotFailedBeforeRequest = true } + // 我方执行时限到点不是"租约丢失":续租请求可能正好被同一个时限打断, + // 此时必须继续走失败收尾,否则任务停在 running、租约过期后被反复重跑。 + deadlineExpired := errors.Is(ctx.Err(), context.DeadlineExceeded) select { case leaseErr := <-leaseLost: - _ = s.log(task.UserID, task.ID, "warn", "任务租约失效,等待其他 worker 恢复", leaseErr.Error()) - return leaseErr + if !deadlineExpired { + _ = s.log(task.UserID, task.ID, "warn", "任务租约失效,等待其他 worker 恢复", leaseErr.Error()) + return leaseErr + } + _ = s.log(task.UserID, task.ID, "warn", "任务执行时限到点,按失败收尾(不视为租约失效)", leaseErr.Error()) default: } decryptedInput, decryptErr := s.decryptTaskInputJSON(task.InputJSON) @@ -297,6 +303,15 @@ func taskFailureMessage(err error) string { return truncateRunes(err.Error(), 2_000) } +// taskLeaseRenewContext 给续租单独一份"不继承父 context 取消/时限"的上下文(仅 5 秒上限)。 +// +// 续租必须比"这一条任务的执行时限"活得更久:父 context 一旦到点,派生的续租 context 会立刻 +// 被取消,续租请求带着 context.Canceled 失败并被误判成"租约失效",任务于是停在 running, +// 租约过期后又被其它 worker 重跑(实测一次上游调用被重跑成三次)。 +func taskLeaseRenewContext(parent context.Context) (context.Context, context.CancelFunc) { + return context.WithTimeout(context.WithoutCancel(parent), 5*time.Second) +} + func taskExecutionTimeoutWithPolicy(taskType string, policy RuntimeTaskPolicy) time.Duration { switch { case strings.HasPrefix(taskType, "canvas_video") || strings.HasPrefix(taskType, "video_"): diff --git a/backend/internal/app/task_worker_test.go b/backend/internal/app/task_worker_test.go new file mode 100644 index 000000000..e7832f8dc --- /dev/null +++ b/backend/internal/app/task_worker_test.go @@ -0,0 +1,41 @@ +package app + +import ( + "context" + "testing" + "time" +) + +// 续租必须比"这条任务的执行时限"活得更久:否则父 context 到点时,续租请求会被自己的 +// 执行时限取消,失败后被误判成"租约失效"(任务停在 running,租约过期后被别的 worker 重跑)。 +func TestTaskLeaseRenewContextSurvivesExecutionDeadline(t *testing.T) { + parent, cancelParent := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancelParent() + <-parent.Done() + if parent.Err() == nil { + t.Fatal("父 context 应当已到点") + } + + renew, cancelRenew := taskLeaseRenewContext(parent) + defer cancelRenew() + if err := renew.Err(); err != nil { + t.Fatalf("父 context 到点后,续租 context 不该被取消:%v", err) + } + deadline, ok := renew.Deadline() + if !ok { + t.Fatal("续租 context 必须保有自己的上限") + } + if remaining := time.Until(deadline); remaining <= 0 || remaining > 5*time.Second { + t.Fatalf("续租上限应当是全新的 5 秒,实际剩余 %v", remaining) + } + + // 父 context 之后再取消,也不能把已经建好的续租 context 一起取消(worker 退出时 + // 续租请求最多自己超时,而不是被"任务已结束"顺手打断)。 + parent2, cancelParent2 := context.WithCancel(context.Background()) + renew2, cancelRenew2 := taskLeaseRenewContext(parent2) + defer cancelRenew2() + cancelParent2() + if err := renew2.Err(); err != nil { + t.Fatalf("父 context 取消不该连带取消续租 context:%v", err) + } +}