441 lines
16 KiB
Go
441 lines
16 KiB
Go
package app
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"log"
|
||
"strings"
|
||
"time"
|
||
|
||
"yingce/backend/internal/model"
|
||
"yingce/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)
|
||
}
|
||
|
||
// wakeTaskDispatcher lets newly persisted work enter execution immediately;
|
||
// the periodic scan remains the cross-process and missed-notification recovery path.
|
||
func (s *Service) wakeTaskDispatcher() {
|
||
if s == nil || s.taskDispatcherWake == nil {
|
||
return
|
||
}
|
||
select {
|
||
case s.taskDispatcherWake <- struct{}{}:
|
||
default:
|
||
}
|
||
}
|
||
|
||
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) {
|
||
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()
|
||
s.wakeTaskDispatcher()
|
||
}()
|
||
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 <-s.taskDispatcherWake:
|
||
dispatch()
|
||
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(), taskExecutionTimeout(task, 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)
|
||
}
|
||
if task.MediaRecoveryJSON != "" {
|
||
result, recoveryErr := s.resumeTaskMedia(ctx, task)
|
||
return s.finishTaskMediaRecovery(task, result, recoveryErr)
|
||
}
|
||
|
||
// Creation and manual retry admission can precede execution by a long time.
|
||
// Re-read the feature policy before preparing or dispatching a model route.
|
||
decryptedInput, err := s.decryptTaskInputJSON(task.InputJSON)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
var admissionInput map[string]any
|
||
if err := json.Unmarshal([]byte(decryptedInput), &admissionInput); err != nil {
|
||
return err
|
||
}
|
||
if err := s.requireCustomChannelsForTaskInput(admissionInput); err != nil {
|
||
// A reclaimed task may already have submitted upstream. Keep the existing
|
||
// review path for that evidence; only an unsent reservation is refundable.
|
||
attempts, readErr := s.repo.RouteAttempts(task.ID, task.RouteRun)
|
||
uncertain := readErr != nil || task.ProviderRequestID != "" || s.taskBilling().BillingFailureRequiresReview(task.BillingOrderID, task.ID, err)
|
||
for _, attempt := range attempts {
|
||
if attempt.DispatchState == "accepted" || attempt.DispatchState == "submission_unknown" {
|
||
uncertain = true
|
||
}
|
||
}
|
||
// Legacy tasks without an explicit new retry run retain their conservative
|
||
// submission handling, as in createDirectTaskAttempt.
|
||
if len(attempts) == 0 && task.RouteRun <= 1 && task.Attempts > 1 {
|
||
uncertain = true
|
||
}
|
||
return terminal.markPreparationFailure(task, "渠道不可用", errors.Join(err, readErr), uncertain, "自定义渠道已关闭，上游请求未发出")
|
||
}
|
||
|
||
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
|
||
latestMedia, readErr := s.repo.Task(task.ID)
|
||
if readErr != nil {
|
||
return readErr
|
||
}
|
||
task.MediaRecoveryJSON, task.MediaStage = latestMedia.MediaRecoveryJSON, latestMedia.MediaStage
|
||
var deliveryFailure *mediaRecoveryError
|
||
if task.MediaRecoveryJSON != "" || errors.As(err, &deliveryFailure) {
|
||
return s.finishTaskMediaRecovery(task, result, 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
|
||
}
|
||
// 续租使用独立 context；执行超时不能覆盖真实租约失效，否则旧 worker
|
||
// 可能在新 worker 接管后继续结算或写入终态。
|
||
deadlineExpired := errors.Is(ctx.Err(), context.DeadlineExceeded)
|
||
select {
|
||
case leaseErr := <-leaseLost:
|
||
_ = s.log(task.UserID, task.ID, "warn", "任务租约失效，等待其他 worker 恢复", leaseErr.Error())
|
||
return leaseErr
|
||
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) || deadlineExpired {
|
||
// 画布 Agent 的单步超时是可恢复事件（运行期会关思考重试同一步），
|
||
// 因此必须与"任务执行超时"区分开，否则只能整轮判死。
|
||
if cloudAgentModelOperation(task) {
|
||
err = errors.New(cloudAgentStepTimeoutError + "，已中止这一步")
|
||
} else {
|
||
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 "任务处理失败"
|
||
}
|
||
message := err.Error()
|
||
if denied := speechResourceDeniedUserMessage(message); denied != "" {
|
||
return denied
|
||
}
|
||
return truncateRunes(message, 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)
|
||
}
|
||
|
||
// taskExecutionTimeout 解析一次任务的执行墙钟：画布 Agent 的单步调用可以配秒级超时
|
||
// （AgentStepTimeoutSeconds），没配时沿用文本任务超时。秒级粒度是必要的——一轮里每一步
|
||
// 都是分钟级的调用，分钟粒度改不动"某一步卡住"的体验。
|
||
// 超时的表现是任务错误里带 cloudAgentStepTimeoutError 标记，运行期据此关思考重试同一步，
|
||
// 而不是把整轮判死（见 cloud_agent_step_timeout.go）。
|
||
func taskExecutionTimeout(task *model.Task, policy RuntimeTaskPolicy) time.Duration {
|
||
if task != nil && cloudAgentModelOperation(task) && policy.AgentStepTimeoutSeconds > 0 {
|
||
return time.Duration(policy.AgentStepTimeoutSeconds) * time.Second
|
||
}
|
||
if task == nil {
|
||
return time.Duration(policy.DefaultTimeoutMinutes) * time.Minute
|
||
}
|
||
return taskExecutionTimeoutWithPolicy(task.Type, policy)
|
||
}
|
||
|
||
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 "任务执行超时，请稍后重试。"
|
||
}
|