fix(tasks): 任务队列 - 执行前重查排队任务的自定义渠道开关 (#682)

Co-authored-by: ddcat666 <26564566+ddcat666@users.noreply.github.com>
2026-10-06 09:55:47 +08:00
parent 9bc255467d
commit f6dd7860aa
4 changed files with 162 additions and 0 deletions
--- a/backend/internal/app/task_queued_channel_test.go
+++ b/backend/internal/app/task_queued_channel_test.go
@@ -0,0 +1,126 @@
+package app
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"infinite-canvas/backend/internal/model"
+)
+
+type queuedChannelExecutionProbe struct {
+	*Service
+	calls int
+}
+
+func (p *queuedChannelExecutionProbe) processTask(context.Context, model.Task) (map[string]interface{}, []map[string]interface{}, error) {
+	p.calls++
+	return nil, nil, errors.New("execution probe reached")
+}
+
+func TestQueuedCustomChannelRecheck(t *testing.T) {
+	for _, tc := range []struct {
+		name                              string
+		retry, enabled, system, uncertain bool
+		dispatch                          string
+	}{
+		{name: "new"}, {name: "retry", retry: true},
+		{name: "enabled", enabled: true}, {name: "system", system: true},
+		{name: "accepted", dispatch: "accepted", uncertain: true},
+		{name: "submission_unknown", dispatch: "submission_unknown", uncertain: true},
+		{name: "billing_review", uncertain: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			s, db := newMediaRecoveryTestService(t)
+			actor := &model.User{ID: "admin", Role: "admin"}
+			features := FeatureAvailability{CustomChannelsEnabled: true, TaskCenterEnabled: true, CreditsEnabled: true}
+			if _, err := s.UpdateFeatureAvailability(actor, features); err != nil {
+				t.Fatal(err)
+			}
+			input := map[string]any{"mode": "text", "config": map[string]any{"baseUrl": "https://provider.example", "apiKey": "test-key", "model": "test"}}
+			if tc.system {
+				input["config"].(map[string]any)["channelId"] = "system-channel"
+			}
+			if err := s.requireCustomChannelsForTaskInput(input); err != nil {
+				t.Fatal(err)
+			}
+			if err := s.protectTaskSecrets(input); err != nil {
+				t.Fatal(err)
+			}
+			encoded, _ := json.Marshal(input)
+			task := &model.Task{ID: newID(), UserID: "user", Type: "canvas_text", Status: model.TaskStatusQueued, InputJSON: string(encoded), RouteRun: 1}
+			if tc.retry {
+				task.Status = model.TaskStatusFailed
+				task.Attempts = 1
+			}
+			if err := db.Create(task).Error; err != nil {
+				t.Fatal(err)
+			}
+			if tc.retry {
+				var err error
+				task, err = s.repo.RetryTaskWithBilling(task.UserID, task, nil, 10)
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+			order := model.BillingOrder{ID: newID(), UserID: task.UserID, TaskID: task.ID, IdempotencyKey: newID(), Status: model.BillingStatusReserved, AmountMicrocredits: 100, ReservedAmountMicrocredits: 100}
+			if tc.uncertain && tc.dispatch == "" {
+				order.Status = model.BillingStatusUncertain
+			}
+			if tc.dispatch != "" {
+				attempt := &model.RouteAttempt{ID: newID(), TaskID: task.ID, RouteRun: task.RouteRun, DispatchState: tc.dispatch, Status: "selected", AttemptNumber: 1}
+				if err := db.Create(attempt).Error; err != nil {
+					t.Fatal(err)
+				}
+			}
+			for _, value := range []any{&order, &model.CreditAccount{UserID: task.UserID, AvailableMicrocredits: 900, ReservedMicrocredits: 100}} {
+				if err := db.Create(value).Error; err != nil {
+					t.Fatal(err)
+				}
+			}
+			if err := db.Model(task).Update("billing_order_id", order.ID).Error; err != nil {
+				t.Fatal(err)
+			}
+			features.CustomChannelsEnabled = tc.enabled
+			if _, err := s.UpdateFeatureAvailability(actor, features); err != nil {
+				t.Fatal(err)
+			}
+			claimed, err := s.repo.ClaimNextTask("worker", time.Minute)
+			if err != nil || claimed == nil {
+				t.Fatalf("claim: %v %v", claimed, err)
+			}
+			probe := &queuedChannelExecutionProbe{Service: s}
+			s.taskRouteExecutor = &taskRouteExecutor{port: probe}
+			err = s.taskWorker().processClaimedTask(claimed, nil)
+			stored, readErr := s.repo.Task(task.ID)
+			if readErr != nil {
+				t.Fatal(readErr)
+			}
+			wantCalls, wantError := 0, "自定义渠道"
+			if tc.enabled || tc.system {
+				wantCalls, wantError = 1, "execution probe reached"
+			}
+			if err == nil || !strings.Contains(stored.Error, wantError) || stored.Status != model.TaskStatusFailed || probe.calls != wantCalls || stored.CompletedAt == nil || stored.LeaseOwner != "" {
+				t.Errorf("queued channel rejection: err=%v status=%s error=%q calls=%d", err, stored.Status, stored.Error, probe.calls)
+			}
+			got, err := s.repo.BillingOrder(order.ID)
+			if err != nil {
+				t.Fatal(err)
+			}
+			var account model.CreditAccount
+			if err := db.First(&account, "user_id = ?", task.UserID).Error; err != nil {
+				t.Fatal(err)
+			}
+			wantStatus, available, reserved := model.BillingStatusRefunded, int64(1000), int64(0)
+			if tc.uncertain {
+				wantStatus, available, reserved = model.BillingStatusUncertain, 900, 100
+			}
+			if got.Status != wantStatus || account.AvailableMicrocredits != available || account.ReservedMicrocredits != reserved {
+				t.Errorf("reservation not refunded: status=%s available=%d reserved=%d", got.Status, account.AvailableMicrocredits, account.ReservedMicrocredits)
+			}
+		})
+	}
+}
--- a/backend/internal/app/task_worker.go
+++ b/backend/internal/app/task_worker.go
@@ -191,6 +191,34 @@ func (w *taskWorkerCoordinator) processClaimedTask(task *model.Task, globalSlot
 		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
--- a/docs/content/docs/backend/task-runtime-stability.mdx
+++ b/docs/content/docs/backend/task-runtime-stability.mdx
@@ -5,6 +5,10 @@ description: 热点回源保护、并发租约、故障边界和上线验收

 # 任务缓存与队列稳定性

+## 排队后的自定义渠道检查
+
+worker 在准备模型路由前重新读取自定义渠道开关，覆盖首次排队和失败后重新入队。排队期间关闭开关的任务以“自定义渠道”不可用的明确错误进入失败终态，不调用渠道；未发出的预扣沿用已有失败退款路径。已接受、提交不明或原本待核对的订单仍进入既有待核对路径，不将可能已发生的费用自动退回。已生成作品的媒体保存恢复不重新生成，不受此入口阻断。
+
 ## 详情读取和日志筛选边界

 任务中心打开详情只读取任务和日志，不在该路径调用画布结果同步。后台同步遇到短剧领域项目任务（`domainProjectId` 等于任务 `projectId`）或没有生成节点关联时跳过画布写回，避免把短剧项目 ID 当作画布 ID 查询。
--- a/docs/content/docs/progress/pending-test.mdx
+++ b/docs/content/docs/progress/pending-test.mdx
@@ -23,6 +23,10 @@ description: 当前项目已实现功能的手工验证清单与未确认边界
 - [ ] 运行 `TestUploadContentImportSocket` 验证 URL 导入与重试；在可解析外部存储域名的环境复测 `TestResolveAccessPolicyMatrix`。
 - [ ] PostgreSQL 上传重试、真实 OSS 附件代理与浏览器附件下载验收。

+## 排队后的自定义渠道检查
+
+- [ ] PostgreSQL 环境验证排队期间关闭自定义渠道：首次与重试任务均停止调用，未发出订单退款，已提交或待核对订单保持待核对。
+
 ## Agent 失败消息合并与展示

 - 已修复：同一运行的 `run_failed` / `error` 与失败状态快照共用一条错误消息；以公开失败原因补全内容，重放、乱序和空原因不覆盖已有详细说明。不同运行即使错误文本相同也分别保留，工具失败与连接错误不合并。