04d8b85eec
Original-Commit: 2b796c403f0c99f3b607b53650b42078964ea753 fix(app): 任务租约 - 续租不再被自己的执行时限打断 续租 context 原来派生自任务执行时限（context.WithTimeout(ctx, 5s)）：执行时限一到， 续租请求立刻被取消，失败后被误判成"租约失效、等待其他 worker 恢复"，任务停在 running， 租约过期后又被别的 worker 重跑（实测一次上游调用被重跑成三次）。 - taskLeaseRenewContext：用 context.WithoutCancel(parent) + 5s，只继承值、不继承取消与时限； - 收尾处若我方执行时限已到点，不再直接返回租约错误，而是继续走失败收尾（真丢失时行为不变）。 新增用例在旧写法下失败、修复后通过。
377 lines
14 KiB
Go
377 lines
14 KiB
Go
package app
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"log"
|
||
"strings"
|
||
"time"
|
||
|
||
"infinite-canvas/backend/internal/model"
|
||
"infinite-canvas/backend/internal/platform"
|
||
)
|
||
|
||
const newAPIChannel2TaskSyncMaxAge = 5 * time.Minute
|
||
|
||
// taskWorkerCoordinator 收敛任务领取、租约维护和执行结果落库，避免 Service 同时承担 worker 生命周期与业务命令。
|
||
type taskWorkerCoordinator struct {
|
||
service *Service
|
||
}
|
||
|
||
const workerSlotLeaseDuration = time.Minute
|
||
|
||
func newTaskWorkerCoordinator(service *Service) *taskWorkerCoordinator {
|
||
return &taskWorkerCoordinator{service: service}
|
||
}
|
||
|
||
func (s *Service) taskWorker() *taskWorkerCoordinator {
|
||
if s.taskWorkerCoordinator != nil {
|
||
return s.taskWorkerCoordinator
|
||
}
|
||
// 部分单元测试直接构造 Service 字面量；延迟创建保持这些测试和内部工具兼容。
|
||
return newTaskWorkerCoordinator(s)
|
||
}
|
||
|
||
func (w *taskWorkerCoordinator) start(ctx context.Context) {
|
||
s := w.service
|
||
s.startTextReplayCleanup(ctx)
|
||
s.startProviderCancellationReconciliation(ctx)
|
||
s.startBillingReviewAudit(ctx)
|
||
s.startAgentMemoryCompactScheduler()
|
||
s.runWorkerLoop(func(ctx context.Context) {
|
||
ticker := time.NewTicker(2 * time.Second)
|
||
defer ticker.Stop()
|
||
for {
|
||
if ctx.Err() != nil {
|
||
return
|
||
}
|
||
if !s.IsDraining() {
|
||
s.advanceCloudAgents()
|
||
}
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
}
|
||
}
|
||
})
|
||
s.runWorkerLoop(func(ctx context.Context) {
|
||
slots := make(chan struct{}, maxChannelConcurrencyLimit)
|
||
dispatch := func() {
|
||
if ctx.Err() != nil || s.IsDraining() {
|
||
return
|
||
}
|
||
setting, err := s.runtimeConcurrencySetting()
|
||
if err != nil {
|
||
log.Printf("task dispatch paused: stage=runtime_policy worker_id=%s error=%v", s.workerID, err)
|
||
return
|
||
}
|
||
workerConcurrency := setting.WorkerConcurrency
|
||
for len(slots) < workerConcurrency {
|
||
globalSlot, acquired, err := s.coordinator.AcquireLease(ctx, "workers", workerConcurrency, workerSlotLeaseDuration)
|
||
if err != nil || !acquired {
|
||
if err != nil {
|
||
log.Printf("task dispatch paused: stage=global_slot worker_id=%s error=%v", s.workerID, err)
|
||
}
|
||
return
|
||
}
|
||
claimCtx, cancelClaim := context.WithTimeout(ctx, 5*time.Second)
|
||
// 每次领取使用独立 owner，防止同一进程内旧执行者恢复后覆盖新执行者。
|
||
task, err := s.repo.WithContext(claimCtx).ClaimNextTask(globalSlot.Token(), 45*time.Second)
|
||
cancelClaim()
|
||
if err != nil || task == nil {
|
||
globalSlot.Release()
|
||
if err != nil {
|
||
log.Printf("task dispatch paused: stage=claim worker_id=%s error=%v", s.workerID, err)
|
||
}
|
||
return
|
||
}
|
||
slots <- struct{}{}
|
||
started := s.runWorkerTask(func() {
|
||
defer func() { <-slots; globalSlot.Release() }()
|
||
if err := w.processClaimedTask(task, globalSlot); err != nil {
|
||
_ = s.log(task.UserID, task.ID, "error", "后台任务处理失败", err.Error())
|
||
}
|
||
})
|
||
if !started {
|
||
<-slots
|
||
globalSlot.Release()
|
||
_ = s.repo.ReleaseTaskLease(task.ID, task.LeaseOwner)
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
dispatch()
|
||
ticker := time.NewTicker(2 * time.Second)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
dispatch()
|
||
}
|
||
}
|
||
})
|
||
}
|
||
|
||
func (w *taskWorkerCoordinator) processNextTask() error {
|
||
s := w.service
|
||
task, err := s.repo.ClaimNextTask(s.workerID+":"+newID(), 45*time.Second)
|
||
if err != nil || task == nil {
|
||
return err
|
||
}
|
||
return w.processClaimedTask(task, nil)
|
||
}
|
||
|
||
func (w *taskWorkerCoordinator) processClaimedTask(task *model.Task, globalSlot *platform.SlotLease) error {
|
||
s := w.service
|
||
terminal := s.terminalCoordinator()
|
||
policyCtx, cancelPolicy := context.WithTimeout(context.Background(), 3*time.Second)
|
||
reader := &Service{repo: s.repo.WithContext(policyCtx)}
|
||
policy, err := reader.RuntimePolicy()
|
||
cancelPolicy()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), taskExecutionTimeoutWithPolicy(task.Type, policy.Task))
|
||
defer cancel()
|
||
leaseDone := make(chan struct{})
|
||
leaseLost := make(chan error, 1)
|
||
taskID, leaseOwner := task.ID, task.LeaseOwner
|
||
go func() {
|
||
ticker := time.NewTicker(15 * time.Second)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-ticker.C:
|
||
renewCtx, cancelRenew := taskLeaseRenewContext(ctx)
|
||
var err error
|
||
if globalSlot != nil {
|
||
err = globalSlot.Renew(renewCtx)
|
||
}
|
||
if err == nil {
|
||
err = s.repo.WithContext(renewCtx).RenewTaskLease(taskID, leaseOwner, 45*time.Second)
|
||
}
|
||
cancelRenew()
|
||
if err != nil {
|
||
leaseLost <- err
|
||
cancel()
|
||
return
|
||
}
|
||
case <-leaseDone:
|
||
return
|
||
}
|
||
}
|
||
}()
|
||
defer close(leaseDone)
|
||
_ = s.log(task.UserID, task.ID, "info", "后端任务开始处理", "")
|
||
s.registerActiveTask(task.ID, cancel)
|
||
defer s.unregisterActiveTask(task.ID)
|
||
// 取消请求可能在任务领取和注册 worker context 之间到达；再次读取终态
|
||
// 可以避免这种极窄窗口仍然向上游发起调用。
|
||
if latest, latestErr := s.repo.Task(task.ID); latestErr == nil && latest.Status == model.TaskStatusCancelled {
|
||
return terminal.handleAlreadyCancelled(*latest)
|
||
}
|
||
|
||
// 时间线转写由本地 whisper.cpp 执行，不经计费与模型渠道路由，
|
||
// 在进入通用生成流程前按类型分叉到独立执行器。
|
||
if task.Type == model.TaskTypeTimelineTranscription {
|
||
return w.processTimelineTranscription(task, ctx)
|
||
}
|
||
if task.Type == model.TaskTypeTimelineRender {
|
||
return w.processTimelineRender(task, ctx)
|
||
}
|
||
|
||
s.markAgentMemoryCompactRunning(*task)
|
||
task.Stage = "调用生成模型"
|
||
task.Progress = 35
|
||
if taskUsesUpstreamReportedProgress(task.Type) {
|
||
// 图片/视频百分比只能来自供应商状态响应。连接和提交阶段只展示文案，
|
||
// 不能再用统一的 35% 冒充真实生成进度。
|
||
task.Stage = "正在连接上游"
|
||
task.Progress = 0
|
||
}
|
||
if err := s.repo.UpdateTaskProgressForLease(task.ID, task.LeaseOwner, task.Stage, task.Progress); err != nil {
|
||
return fmt.Errorf("更新任务进度失败，任务暂未调用上游：%w", err)
|
||
}
|
||
routeAttempt, err := s.beginTaskRouteAttempt(task)
|
||
if err != nil {
|
||
return terminal.markPreparationFailure(task, "路由准备失败", err, isRouteDispatchUncertain(err), "路由准备失败，上游请求未发出")
|
||
}
|
||
if err := s.taskBilling().MarkBillingRunning(task.BillingOrderID); err != nil {
|
||
return terminal.markPreparationFailure(task, "计费准备失败", err, false, "计费准备失败，上游请求未发出")
|
||
}
|
||
routeResult, stateErr := s.routeExecutor().execute(ctx, task, routeAttempt)
|
||
if stateErr != nil {
|
||
return stateErr
|
||
}
|
||
select {
|
||
case leaseErr := <-leaseLost:
|
||
return fmt.Errorf("任务租约失效，停止保存上游结果：%w", leaseErr)
|
||
default:
|
||
}
|
||
result, canvasOps, err := routeResult.result, routeResult.canvasOps, routeResult.err
|
||
providerSucceeded := routeResult.providerSucceeded
|
||
if err == nil {
|
||
result, err = s.persistGeneratedMediaResult(task.UserID, result)
|
||
}
|
||
if err == nil {
|
||
_, err = s.finalizeCharacterTurnaroundTask(*task, result)
|
||
}
|
||
if err != nil {
|
||
channelSlotFailedBeforeRequest := false
|
||
if code, _ := ChannelSlotFailureDetails(err); code != "" {
|
||
channelSlotFailedBeforeRequest = true
|
||
}
|
||
// 我方执行时限到点不是"租约丢失"：续租请求可能正好被同一个时限打断，
|
||
// 此时必须继续走失败收尾，否则任务停在 running、租约过期后被反复重跑。
|
||
deadlineExpired := errors.Is(ctx.Err(), context.DeadlineExceeded)
|
||
select {
|
||
case leaseErr := <-leaseLost:
|
||
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)
|
||
if decryptErr == nil && s.shouldDeferVideoProviderTask(*task, decryptedInput, err) {
|
||
stage := "后台仍在生成"
|
||
message := "前台等待结束，上游视频仍在生成，将继续回查原任务"
|
||
var pendingErr providerStatePendingError
|
||
if errors.As(err, &pendingErr) {
|
||
stage = "等待上游任务同步"
|
||
message = "上游任务状态暂未同步，将继续回查原任务"
|
||
}
|
||
if deferErr := s.repo.DeferRunningTaskForProviderPoll(task.ID, task.LeaseOwner, stage, 15*time.Second); deferErr != nil {
|
||
return deferErr
|
||
}
|
||
_ = s.log(task.UserID, task.ID, "info", message, task.PollStage)
|
||
return nil
|
||
}
|
||
if newAPIChannel2TaskSyncExpired(*task, err, time.Now()) {
|
||
err = errors.New("上游任务长时间未同步，已停止自动查询，请确认渠道任务状态后重试。")
|
||
}
|
||
if errors.Is(err, context.DeadlineExceeded) {
|
||
err = errors.New(taskTimeoutMessage(task.Type))
|
||
}
|
||
s.noteAgentMemoryCompactTask(*task, nil, err)
|
||
return terminal.handleExecutionFailure(task, err, providerSucceeded, channelSlotFailedBeforeRequest)
|
||
}
|
||
latest, err := s.repo.Task(task.ID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if latest.Status == model.TaskStatusCancelled {
|
||
s.noteAgentMemoryCompactTask(*task, nil, errors.New("压缩任务已取消"))
|
||
return terminal.handleCancelledResult(*latest)
|
||
}
|
||
resultJSON, err := json.Marshal(result)
|
||
if err != nil {
|
||
s.noteAgentMemoryCompactTask(*task, nil, err)
|
||
_, terminalErr := terminal.handleResultPersistenceFailure(task, fmt.Errorf("序列化任务结果失败：%w", err))
|
||
return terminalErr
|
||
}
|
||
opsJSON, err := json.Marshal(canvasOps)
|
||
if err != nil {
|
||
s.noteAgentMemoryCompactTask(*task, nil, err)
|
||
_, terminalErr := terminal.handleResultPersistenceFailure(task, fmt.Errorf("序列化画布操作失败：%w", err))
|
||
return terminalErr
|
||
}
|
||
if err := s.saveTaskCompletionWithinStorageQuota(task, resultJSON, opsJSON, len(canvasOps) > 0); err != nil {
|
||
s.noteAgentMemoryCompactTask(*task, nil, err)
|
||
_, terminalErr := terminal.handleResultPersistenceFailure(task, err)
|
||
return terminalErr
|
||
}
|
||
s.noteAgentMemoryCompactTask(*task, result, nil)
|
||
return terminal.handleSuccess(task)
|
||
}
|
||
|
||
func taskUsesUpstreamReportedProgress(taskType string) bool {
|
||
return taskType == "canvas_image" || taskType == "canvas_video" || strings.HasPrefix(taskType, "video_")
|
||
}
|
||
|
||
func taskFailureMessage(err error) string {
|
||
if err == nil {
|
||
return "任务处理失败"
|
||
}
|
||
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_"):
|
||
return max(time.Duration(policy.VideoTimeoutMinutes)*time.Minute, 5*time.Minute)
|
||
case strings.HasPrefix(taskType, "canvas_image"):
|
||
return time.Duration(policy.ImageTimeoutMinutes) * time.Minute
|
||
case strings.HasPrefix(taskType, "canvas_audio"):
|
||
return time.Duration(policy.AudioTimeoutMinutes) * time.Minute
|
||
case strings.HasPrefix(taskType, "canvas_text"):
|
||
return time.Duration(policy.TextTimeoutMinutes) * time.Minute
|
||
case taskType == model.TaskTypeTimelineTranscription:
|
||
// 本地转写受媒体时长影响，给出独立宽松超时。
|
||
return 20 * time.Minute
|
||
case taskType == model.TaskTypeTimelineRender:
|
||
// 渲染是整条时间线的重编码，耗时随长度线性增长。
|
||
return 60 * time.Minute
|
||
default:
|
||
return time.Duration(policy.DefaultTimeoutMinutes) * time.Minute
|
||
}
|
||
}
|
||
|
||
func (s *Service) shouldDeferVideoProviderTask(task model.Task, decryptedInput string, err error) bool {
|
||
providerRequestID := strings.TrimSpace(task.ProviderRequestID)
|
||
if providerRequestID == "" || (!strings.HasPrefix(task.Type, "canvas_video") && !strings.HasPrefix(task.Type, "video_")) {
|
||
return false
|
||
}
|
||
deferSignal := errors.Is(err, context.DeadlineExceeded)
|
||
var pendingErr providerStatePendingError
|
||
if errors.As(err, &pendingErr) {
|
||
deferSignal = strings.TrimSpace(pendingErr.TaskID) == providerRequestID && !newAPIChannel2TaskSyncExpired(task, err, time.Now())
|
||
}
|
||
if !deferSignal {
|
||
return false
|
||
}
|
||
var input canvasGenerationInput
|
||
if json.Unmarshal([]byte(decryptedInput), &input) != nil {
|
||
return false
|
||
}
|
||
resolved, resolveErr := s.resolveProviderConfig(input.Config)
|
||
return resolveErr == nil && resolved.InterfaceType == string(model.ChannelInterfaceNewAPIChannel2)
|
||
}
|
||
|
||
func newAPIChannel2TaskSyncExpired(task model.Task, err error, now time.Time) bool {
|
||
var pendingErr providerStatePendingError
|
||
if !errors.As(err, &pendingErr) || strings.TrimSpace(pendingErr.TaskID) == "" || strings.TrimSpace(pendingErr.TaskID) != strings.TrimSpace(task.ProviderRequestID) {
|
||
return false
|
||
}
|
||
if task.StartedAt == nil {
|
||
return true
|
||
}
|
||
return !now.Before(task.StartedAt.Add(newAPIChannel2TaskSyncMaxAge))
|
||
}
|
||
|
||
func taskTimeoutMessage(taskType string) string {
|
||
if strings.HasPrefix(taskType, "canvas_video") || strings.HasPrefix(taskType, "video_") {
|
||
return "视频生成等待超时，请稍后到任务中心查看或重试。"
|
||
}
|
||
if strings.HasPrefix(taskType, "canvas_image") {
|
||
return "图片生成等待超时，请稍后重试。"
|
||
}
|
||
return "任务执行超时，请稍后重试。"
|
||
}
|