diff --git a/backend/internal/app/admin.go b/backend/internal/app/admin.go index 27696ca6b..51bff5aba 100644 --- a/backend/internal/app/admin.go +++ b/backend/internal/app/admin.go @@ -662,10 +662,18 @@ func (s *Service) LogAPICall(log model.ApiCallLog) error { var nextPollAt *time.Time if stage == "create" && log.Status == model.ApiCallStatusSucceeded && log.ProviderRequestID != "" { stage = "accepted" - next := time.Now().Add(2 * time.Second) + delay := 2 * time.Second + if log.Capability == "video" { + delay = defaultVideoPollInterval + } + next := time.Now().Add(delay) nextPollAt = &next } else if stage == "poll" { - next := time.Now().Add(5 * time.Second) + delay := 5 * time.Second + if log.Capability == "video" { + delay = defaultVideoPollInterval + } + next := time.Now().Add(delay) nextPollAt = &next } if err := s.repo.UpdateTaskProviderState(log.TaskID, log.ProviderRequestID, stage, nextPollAt); err != nil { diff --git a/backend/internal/app/protocol_runtime_test.go b/backend/internal/app/protocol_runtime_test.go index f13111f81..63c932f85 100644 --- a/backend/internal/app/protocol_runtime_test.go +++ b/backend/internal/app/protocol_runtime_test.go @@ -337,7 +337,11 @@ func TestDeclarativeProtocolRuntimeExecutesCreatePollAndDownload(t *testing.T) { config := providerConfig{BaseURL: server.URL + "/v1", APIKey: "key", Model: "test-model", APIFormat: "openai", InterfaceType: "test-declarative-video-runtime"} ctx := context.Background() ctx = withProtocolRegistry(ctx, center.registrySnapshot()) - result, err := runDeclarativeProtocolTask(ctx, canvasGenerationInput{Mode: "video", Prompt: "a clip", Config: config}) + adapter, ok := declarativeProtocolAdapterForContext(ctx, config.InterfaceType) + if !ok { + t.Fatal("declarative adapter is unavailable") + } + result, err := runProtocolAdapterTaskWithPolicy(ctx, canvasGenerationInput{Mode: "video", Prompt: "a clip", Config: config}, adapter, fastVideoPollPolicy()) if err != nil { t.Fatal(err) } @@ -346,6 +350,113 @@ func TestDeclarativeProtocolRuntimeExecutesCreatePollAndDownload(t *testing.T) { } } +func TestDeclarativeProtocolPollingPolicyOnlyChangesVideoBehavior(t *testing.T) { + video := declarativeProtocolPollPolicy("video") + if video.InitialDelay != defaultVideoPollInterval || !video.RetryTransient { + t.Fatalf("video polling policy = %#v, want delayed resilient polling", video) + } + + image := declarativeProtocolPollPolicy("image") + if image.InitialDelay != 0 || image.Interval != 2500*time.Millisecond || image.RetryTransient { + t.Fatalf("image polling policy = %#v, want legacy immediate fail-fast polling", image) + } +} + +func TestDeclarativeProtocolPollRecoversFromTransientGatewayFailure(t *testing.T) { + allowLoopbackProviderTest(t) + adapter, err := protocol.LoadManifest([]byte(`{ + "apiVersion":"yingce.plugin/v1", + "id":"test-declarative-video-retry","version":"1.0.0","name":"Test Declarative Video Retry","author":"Test","documentation":"# Test", + "contributes":{"providers":[{"id":"test-declarative-video-retry","label":"Test Declarative Video Retry","capabilities":["video"],"scopes":["canvas"],"create":{"method":"POST","path":"/tasks","fields":{"model":"request.model"}},"poll":{"method":"GET","path":"/tasks/{{taskId}}"},"response":{"taskIdPaths":["id"],"statusPaths":["status"],"resultPaths":["video_url"],"resultKind":"video"}}]} + }`)) + if err != nil { + t.Fatal(err) + } + createCalls := 0 + pollCalls := 0 + var server *httptest.Server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v1/tasks": + createCalls++ + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"provider-task-1","status":"pending"}`)) + case "/v1/tasks/provider-task-1": + pollCalls++ + w.Header().Set("Content-Type", "application/json") + if pollCalls == 1 { + w.WriteHeader(http.StatusBadGateway) + _, _ = w.Write([]byte(`{"message":"temporary gateway failure"}`)) + return + } + _, _ = w.Write([]byte(`{"id":"provider-task-1","status":"succeeded","video_url":"` + server.URL + `/media.mp4"}`)) + case "/media.mp4": + w.Header().Set("Content-Type", "video/mp4") + _, _ = w.Write([]byte("video")) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + config := providerConfig{BaseURL: server.URL + "/v1", APIKey: "key", Model: "video-model", InterfaceType: "test-declarative-video-retry"} + ctx := context.Background() + result, err := runProtocolAdapterTaskWithPolicy(ctx, canvasGenerationInput{Mode: "video", Prompt: "a clip", Config: config}, adapter, fastVideoPollPolicy()) + if err != nil { + t.Fatal(err) + } + if result["mode"] != "video" || createCalls != 1 || pollCalls != 2 { + t.Fatalf("result = %#v, create calls = %d, poll calls = %d", result, createCalls, pollCalls) + } +} + +func TestDeclarativeProtocolRetriesResultDownloadWithoutRepolling(t *testing.T) { + allowLoopbackProviderTest(t) + adapter, err := protocol.LoadManifest([]byte(`{ + "apiVersion":"yingce.plugin/v1", + "id":"test-declarative-download-retry","version":"1.0.0","name":"Test Declarative Download Retry","author":"Test","documentation":"# Test", + "contributes":{"providers":[{"id":"test-declarative-download-retry","label":"Test Declarative Download Retry","capabilities":["video"],"scopes":["canvas"],"create":{"method":"POST","path":"/tasks","fields":{"model":"request.model"}},"poll":{"method":"GET","path":"/tasks/{{taskId}}"},"response":{"taskIdPaths":["id"],"statusPaths":["status"],"resultPaths":["video_url"],"resultKind":"video"}}]} + }`)) + if err != nil { + t.Fatal(err) + } + pollCalls := 0 + downloadCalls := 0 + var server *httptest.Server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v1/tasks": + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"provider-task-1","status":"pending"}`)) + case "/v1/tasks/provider-task-1": + pollCalls++ + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"provider-task-1","status":"succeeded","video_url":"` + server.URL + `/media.mp4"}`)) + case "/media.mp4": + downloadCalls++ + if downloadCalls == 1 { + w.WriteHeader(http.StatusBadGateway) + return + } + w.Header().Set("Content-Type", "video/mp4") + _, _ = w.Write([]byte("video")) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + config := providerConfig{BaseURL: server.URL + "/v1", APIKey: "key", Model: "video-model", InterfaceType: "test-declarative-download-retry"} + ctx := context.Background() + result, err := runProtocolAdapterTaskWithPolicy(ctx, canvasGenerationInput{Mode: "video", Prompt: "a clip", Config: config}, adapter, fastVideoPollPolicy()) + if err != nil { + t.Fatal(err) + } + if result["mode"] != "video" || pollCalls != 1 || downloadCalls != 2 { + t.Fatalf("result = %#v, poll calls = %d, download calls = %d", result, pollCalls, downloadCalls) + } +} + func TestDeclarativeProtocolRuntimeGeneratesPerCreateIdempotencyKey(t *testing.T) { allowLoopbackProviderTest(t) manifest := []byte(`{ @@ -395,8 +506,12 @@ func TestDeclarativeProtocolRuntimeGeneratesPerCreateIdempotencyKey(t *testing.T config := providerConfig{BaseURL: server.URL + "/v1", APIKey: "key", Model: "test-model", APIFormat: "openai", InterfaceType: "test-idempotency-key-runtime"} ctx := context.Background() ctx = withProtocolRegistry(ctx, center.registrySnapshot()) + adapter, ok := declarativeProtocolAdapterForContext(ctx, config.InterfaceType) + if !ok { + t.Fatal("declarative adapter is unavailable") + } for i := 0; i < 2; i++ { - result, err := runDeclarativeProtocolTask(ctx, canvasGenerationInput{Mode: "video", Prompt: "a clip", Config: config}) + result, err := runProtocolAdapterTaskWithPolicy(ctx, canvasGenerationInput{Mode: "video", Prompt: "a clip", Config: config}, adapter, fastVideoPollPolicy()) if err != nil { t.Fatal(err) } @@ -441,12 +556,7 @@ func TestDeclarativeNewAPIChannel2TaskNotExistRetry(t *testing.T) { config := providerConfig{BaseURL: server.URL + "/v1", APIKey: "key", Model: "video-model", InterfaceType: "newapi-channel-2"} ctx := context.Background() - result, err := runProtocolAdapterTaskWithTiming(ctx, canvasGenerationInput{Mode: "video", Prompt: "a clip", Config: config}, adapter, protocolPollTiming{ - InitialDelay: time.Millisecond, - PollInterval: time.Millisecond, - TaskNotExistWindow: time.Second, - TaskNotExistRetryDelays: []time.Duration{time.Millisecond, time.Millisecond, time.Millisecond, time.Millisecond}, - }) + result, err := runProtocolAdapterTaskWithPolicy(ctx, canvasGenerationInput{Mode: "video", Prompt: "a clip", Config: config}, adapter, fastVideoPollPolicy()) if err != nil { t.Fatal(err) } @@ -479,43 +589,34 @@ func TestDeclarativeNewAPIChannel2TaskNotExistExhaustion(t *testing.T) { config := providerConfig{BaseURL: server.URL + "/v1", APIKey: "key", Model: "video-model", InterfaceType: "newapi-channel-2"} ctx := context.Background() - _, err := runProtocolAdapterTaskWithTiming(ctx, canvasGenerationInput{Mode: "video", Prompt: "a clip", Config: config}, adapter, protocolPollTiming{ - InitialDelay: time.Millisecond, - PollInterval: time.Millisecond, - TaskNotExistWindow: time.Second, - TaskNotExistRetryDelays: []time.Duration{time.Millisecond, time.Millisecond, time.Millisecond, time.Millisecond}, - }) - var pendingErr providerStatePendingError + _, err := runProtocolAdapterTaskWithPolicy(ctx, canvasGenerationInput{Mode: "video", Prompt: "a clip", Config: config}, adapter, fastVideoPollPolicy()) var httpErr providerHTTPError - if !errors.As(err, &pendingErr) || pendingErr.TaskID != "provider-task-1" || !errors.As(err, &httpErr) { - t.Fatalf("error = %#v, want typed provider pending error wrapping HTTP error", err) + if !errors.As(err, &httpErr) || httpErr.StatusCode != http.StatusBadRequest { + t.Fatalf("error = %#v, want provider HTTP error", err) } - if createCalls != 1 || pollCalls != 5 { - t.Fatalf("create calls = %d, poll calls = %d, want 1 and 5", createCalls, pollCalls) + if createCalls != 1 || pollCalls != 3 { + t.Fatalf("create calls = %d, poll calls = %d, want 1 and 3", createCalls, pollCalls) } } -func TestDeclarativeNewAPIChannel2TaskNotExistStrictClassification(t *testing.T) { +func TestProviderTaskNotReadyStrictClassification(t *testing.T) { tests := []struct { - name string - interfaceType string - taskID string - err error - want bool + name string + err providerHTTPError + want bool }{ - {name: "error code", interfaceType: "newapi-channel-2", taskID: "task-1", err: providerHTTPError{StatusCode: 400, Body: `{"code":"task_not_exist"}`}, want: true}, - {name: "error message", interfaceType: "newapi-channel-2", taskID: "task-1", err: providerHTTPError{StatusCode: 400, Body: `{"message":"task_not_exist"}`}, want: true}, - {name: "other interface", interfaceType: "newapi-channel-1", taskID: "task-1", err: providerHTTPError{StatusCode: 400, Body: `{"code":"task_not_exist"}`}}, - {name: "missing task id", interfaceType: "newapi-channel-2", err: providerHTTPError{StatusCode: 400, Body: `{"code":"task_not_exist"}`}}, - {name: "other status", interfaceType: "newapi-channel-2", taskID: "task-1", err: providerHTTPError{StatusCode: 404, Body: `{"code":"task_not_exist"}`}}, - {name: "unstructured body", interfaceType: "newapi-channel-2", taskID: "task-1", err: providerHTTPError{StatusCode: 400, Body: "task_not_exist"}}, - {name: "partial match", interfaceType: "newapi-channel-2", taskID: "task-1", err: providerHTTPError{StatusCode: 400, Body: `{"code":"task_not_exist_later"}`}}, - {name: "other provider error", interfaceType: "newapi-channel-2", taskID: "task-1", err: providerHTTPError{StatusCode: 400, Body: `{"code":"invalid_parameter"}`}}, + {name: "error code", err: providerHTTPError{StatusCode: 400, Body: `{"code":"task_not_exist"}`}, want: true}, + {name: "error message", err: providerHTTPError{StatusCode: 400, Body: `{"message":"task_not_exist"}`}, want: true}, + {name: "not found spelling", err: providerHTTPError{StatusCode: 404, Body: `{"code":"task_not_found"}`}, want: true}, + {name: "other status", err: providerHTTPError{StatusCode: 502, Body: `{"code":"task_not_exist"}`}}, + {name: "unstructured body", err: providerHTTPError{StatusCode: 400, Body: "task_not_exist"}}, + {name: "partial match", err: providerHTTPError{StatusCode: 400, Body: `{"code":"task_not_exist_later"}`}}, + {name: "other provider error", err: providerHTTPError{StatusCode: 400, Body: `{"code":"invalid_parameter"}`}}, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { - if got := isNewAPIChannel2TaskNotReady(test.interfaceType, test.taskID, test.err); got != test.want { - t.Fatalf("isNewAPIChannel2TaskNotReady() = %v, want %v", got, test.want) + if got := isProviderTaskNotReadyError(test.err); got != test.want { + t.Fatalf("isProviderTaskNotReadyError() = %v, want %v", got, test.want) } }) } diff --git a/backend/internal/app/provider.go b/backend/internal/app/provider.go index 77d7b378d..2f1f7b2d9 100644 --- a/backend/internal/app/provider.go +++ b/backend/internal/app/provider.go @@ -98,7 +98,7 @@ type providerConfig struct { } const providerHTTPTimeout = 5 * time.Minute -const videoPollTimeout = 30 * time.Minute +const videoPollTimeout = time.Hour const maxProviderResponseBytes int64 = 64 << 20 type providerMedia struct { @@ -143,6 +143,19 @@ type providerHTTPError struct { RetryAfter time.Duration } +type providerResponseDecodeError struct { + Err error +} + +func (e providerResponseDecodeError) Error() string { return e.Err.Error() } +func (e providerResponseDecodeError) Unwrap() error { return e.Err } + +type providerCircuitOpenError struct{} + +func (providerCircuitOpenError) Error() string { + return "当前渠道连续失败，已暂时熔断，请稍后重试" +} + type providerStatePendingError struct { TaskID string Cause error diff --git a/backend/internal/app/provider_http_client.go b/backend/internal/app/provider_http_client.go index 4ceb52e82..5a7370aa1 100644 --- a/backend/internal/app/provider_http_client.go +++ b/backend/internal/app/provider_http_client.go @@ -188,10 +188,10 @@ func doJSON(req *http.Request, target interface{}) error { return err } if !strings.Contains(mimeType, "json") && !json.Valid(data) { - return fmt.Errorf("接口返回非 JSON 内容：%s", mimeType) + return providerResponseDecodeError{Err: fmt.Errorf("接口返回非 JSON 内容：%s", mimeType)} } if err := json.Unmarshal(data, target); err != nil { - return err + return providerResponseDecodeError{Err: err} } if payload, ok := target.(*imageResponse); ok { if payload.Error != nil && payload.Error.Message != "" { @@ -247,7 +247,7 @@ func doBinaryWithConsumer(req *http.Request, onChunk func(string, []byte)) ([]by return nil, "", fmt.Errorf("读取渠道熔断状态失败：%w", err) } if open { - return nil, "", errors.New("当前渠道连续失败，已暂时熔断，请稍后重试") + return nil, "", providerCircuitOpenError{} } slotID := channelID if slotID == "" { diff --git a/backend/internal/app/provider_protocol.go b/backend/internal/app/provider_protocol.go index 27a53e45f..c4b01106b 100644 --- a/backend/internal/app/provider_protocol.go +++ b/backend/internal/app/provider_protocol.go @@ -22,7 +22,6 @@ import ( "strings" "time" - "infinite-canvas/backend/internal/model" "infinite-canvas/backend/internal/protocol" "github.com/google/uuid" @@ -40,34 +39,28 @@ func runDeclarativeProtocolTask(ctx context.Context, input canvasGenerationInput return runProtocolAdapterTask(ctx, input, adapter) } -type protocolPollTiming struct { - InitialDelay time.Duration - PollInterval time.Duration - TaskNotExistWindow time.Duration - TaskNotExistMaxMisses int - TaskNotExistRetryDelays []time.Duration -} - -var defaultProtocolPollTiming = protocolPollTiming{ - InitialDelay: 2 * time.Second, - PollInterval: 2500 * time.Millisecond, - TaskNotExistWindow: 15 * time.Second, - TaskNotExistMaxMisses: 5, - TaskNotExistRetryDelays: []time.Duration{2 * time.Second, 3 * time.Second, 5 * time.Second, 5 * time.Second}, -} - func runProtocolAdapterTask(ctx context.Context, input canvasGenerationInput, adapter protocol.Adapter) (map[string]interface{}, error) { - return runProtocolAdapterTaskWithTiming(ctx, input, adapter, defaultProtocolPollTiming) + return runProtocolAdapterTaskWithPolicy(ctx, input, adapter, declarativeProtocolPollPolicy(input.Mode)) } -func runProtocolAdapterTaskWithTiming(ctx context.Context, input canvasGenerationInput, adapter protocol.Adapter, timing protocolPollTiming) (map[string]interface{}, error) { +func declarativeProtocolPollPolicy(mode string) videoPollPolicy { + if mode == "video" { + return defaultVideoPollPolicy() + } + return videoPollPolicy{ + Interval: 2500 * time.Millisecond, + MaxNotFoundMisses: 1, + MaxDownloadTries: 3, + Sleep: sleepContext, + } +} + +func runProtocolAdapterTaskWithPolicy(ctx context.Context, input canvasGenerationInput, adapter protocol.Adapter, policy videoPollPolicy) (map[string]interface{}, error) { // create、poll、download 是同一个外部任务的三个阶段：任一阶段失败都向上返回真实错误， // 不把“已提交但结果未知”伪装成成功，也不把下载失败降级成空结果。 request := protocolRequestFromInput(input) taskID := resumedProviderRequestID(ctx) var created protocol.CreateResult - createdProviderTask := false - syncWindowStartedAt := time.Now() if taskID == "" { // 幂等键只存在于宿主请求元数据中，声明式插件可以把它映射到 Header， // 但不能把宿主控制字段泄漏到供应商 JSON body。恢复已有 taskID 时不会进入 create 分支。 @@ -100,71 +93,38 @@ func runProtocolAdapterTaskWithTiming(ctx context.Context, input canvasGeneratio return nil, protocolResultError(created.Message, taskID) } if created.Status == protocol.StatusSucceeded { - return finishProtocolAdapterResult(ctx, input, adapter, request, taskID, created.Result) + return finishProtocolAdapterResult(ctx, input, adapter, request, taskID, created.Result, policy) } if taskID == "" { return nil, errors.New("声明式协议创建请求没有返回任务 ID") } - createdProviderTask = true - syncWindowStartedAt = time.Now() - } - if createdProviderTask && input.Config.InterfaceType == string(model.ChannelInterfaceNewAPIChannel2) { - if err := sleepContext(ctx, timing.InitialDelay); err != nil { - return nil, err - } } - taskNotExistMisses := 0 - for deadline := providerPollingDeadline(ctx); time.Now().Before(deadline); { + return runVideoPollLoop(ctx, taskID, policy, func(ctx context.Context) (videoPollOutcome, error) { spec, err := adapter.BuildPoll(ctx, protocol.PollContext{BaseURL: input.Config.BaseURL, Model: request.Model, Request: request, TaskID: taskID}) if err != nil { - return nil, err + return videoPollOutcome{}, err } body, err := executeProtocolRequest(withProviderRequestKind(ctx, "poll"), input.Config, spec) if err != nil { - if isNewAPIChannel2TaskNotReady(input.Config.InterfaceType, taskID, err) { - // NewAPI Channel 2 创建后存在短暂同步窗口。窗口内的 task_not_exist 只表示尚未可查； - // 超过窗口仍不可查时返回 pending error，让上层保留任务 ID，而不是误判成功或失败。 - taskNotExistMisses++ - maxMisses := timing.TaskNotExistMaxMisses - if maxMisses <= 0 { - maxMisses = len(timing.TaskNotExistRetryDelays) + 1 - } - if time.Since(syncWindowStartedAt) >= timing.TaskNotExistWindow || taskNotExistMisses >= maxMisses { - return nil, providerStatePendingError{TaskID: taskID, Cause: err} - } - delayIndex := min(taskNotExistMisses-1, len(timing.TaskNotExistRetryDelays)-1) - if delayIndex >= 0 { - if sleepErr := sleepContext(ctx, timing.TaskNotExistRetryDelays[delayIndex]); sleepErr != nil { - return nil, sleepErr - } - } - continue - } - return nil, err + return videoPollOutcome{}, err } - taskNotExistMisses = 0 state, err := adapter.ParsePoll(ctx, protocol.PollContext{BaseURL: input.Config.BaseURL, Model: request.Model, Request: request, TaskID: taskID}, body) if err != nil { - return nil, err + return videoPollOutcome{}, err } if state.TaskID != "" { taskID = state.TaskID } switch state.Status { case protocol.StatusSucceeded: - return finishProtocolAdapterResult(ctx, input, adapter, request, taskID, state.Result) + result, err := finishProtocolAdapterResult(ctx, input, adapter, request, taskID, state.Result, policy) + return videoPollOutcome{Done: err == nil, Result: result}, err case protocol.StatusFailed, protocol.StatusCancelled: - return nil, protocolResultError(state.Message, taskID) + return videoPollOutcome{}, protocolResultError(state.Message, taskID) } - if err := sleepContext(ctx, timing.PollInterval); err != nil { - return nil, err - } - } - if err := ctx.Err(); err != nil { - return nil, err - } - return nil, fmt.Errorf("声明式协议任务超时（任务 %s）", taskID) + return videoPollOutcome{}, nil + }) } func extractProviderTaskID(body []byte) (string, error) { @@ -188,22 +148,6 @@ func extractProviderTaskID(body []byte) (string, error) { return "", err } -func isNewAPIChannel2TaskNotReady(interfaceType string, taskID string, err error) bool { - if strings.TrimSpace(interfaceType) != string(model.ChannelInterfaceNewAPIChannel2) || strings.TrimSpace(taskID) == "" { - return false - } - var httpErr providerHTTPError - if !errors.As(err, &httpErr) || httpErr.StatusCode != http.StatusBadRequest { - return false - } - var payload map[string]any - if json.Unmarshal([]byte(httpErr.Body), &payload) != nil { - return false - } - code, message := providerFailureDetails(payload) - return strings.EqualFold(strings.TrimSpace(code), "task_not_exist") || strings.EqualFold(strings.TrimSpace(message), "task_not_exist") -} - // queryProtocolAdapterVideoTask 只读取一次已有的声明式 Provider 任务。 // 人工恢复走此路径，因此检查本地失败任务时绝不会创建第二个计费任务。 func queryProtocolAdapterVideoTask(ctx context.Context, input canvasGenerationInput, adapter protocol.Adapter, taskID string) (map[string]interface{}, string, error) { @@ -224,7 +168,7 @@ func queryProtocolAdapterVideoTask(ctx context.Context, input canvasGenerationIn providerStatus := string(state.Status) switch state.Status { case protocol.StatusSucceeded: - result, err := finishProtocolAdapterResult(ctx, input, adapter, request, taskID, state.Result) + result, err := finishProtocolAdapterResult(ctx, input, adapter, request, taskID, state.Result, defaultVideoPollPolicy()) return result, providerStatus, err case protocol.StatusFailed, protocol.StatusCancelled: return nil, providerStatus, protocolResultError(state.Message, taskID) @@ -803,7 +747,7 @@ func appendProtocolQuery(rawURL string, values map[string][]string) (string, err return parsed.String(), nil } -func finishProtocolResult(ctx context.Context, config providerConfig, mode string, result *protocol.Result) (map[string]interface{}, error) { +func finishProtocolResult(ctx context.Context, config providerConfig, mode string, taskID string, result *protocol.Result, pollPolicy videoPollPolicy) (map[string]interface{}, error) { if result == nil { return nil, errors.New("声明式协议已完成但没有返回结果") } @@ -830,7 +774,16 @@ func finishProtocolResult(ctx context.Context, config providerConfig, mode strin } items := make([]interface{}, 0, len(references)) for _, reference := range references { - data, mimeType, err := protocolMediaBytes(ctx, config, reference) + var data []byte + var mimeType string + var err error + if mode == "video" { + data, mimeType, err = runVideoDownload(ctx, taskID, pollPolicy, func(ctx context.Context) ([]byte, string, error) { + return protocolMediaBytesOnce(ctx, config, reference) + }) + } else { + data, mimeType, err = protocolMediaBytes(ctx, config, reference) + } if err != nil { return nil, err } @@ -850,20 +803,29 @@ func finishProtocolResult(ctx context.Context, config providerConfig, mode strin // finishProtocolAdapterResult 优先消费 create/poll 已返回的内联或 URL 结果，只有插件明确声明独立结果端点时才下载。 // 空响应或下载失败必须向上失败，不能生成伪素材；application/octet-stream 仅表示传输层未知类型， // 不会把未知内容伪装成具体图片、视频或音频 MIME。 -func finishProtocolAdapterResult(ctx context.Context, input canvasGenerationInput, adapter protocol.Adapter, request protocol.GenerationRequest, taskID string, result *protocol.Result) (map[string]interface{}, error) { +func finishProtocolAdapterResult(ctx context.Context, input canvasGenerationInput, adapter protocol.Adapter, request protocol.GenerationRequest, taskID string, result *protocol.Result, pollPolicy videoPollPolicy) (map[string]interface{}, error) { if protocolResultHasOutput(input.Mode, result) { - return finishProtocolResult(ctx, input.Config, input.Mode, result) + return finishProtocolResult(ctx, input.Config, input.Mode, taskID, result, pollPolicy) } resultAdapter, ok := adapter.(protocol.ResultAdapter) capability, hasCapability := adapter.(protocol.ResultCapability) if !ok || !hasCapability || !capability.ResultAvailable() { - return finishProtocolResult(ctx, input.Config, input.Mode, result) + return finishProtocolResult(ctx, input.Config, input.Mode, taskID, result, pollPolicy) } spec, err := resultAdapter.BuildResult(ctx, protocol.PollContext{BaseURL: input.Config.BaseURL, Model: request.Model, Request: request, TaskID: taskID}) if err != nil { return nil, err } - data, mimeType, err := executeProtocolBinaryRequest(withProviderRequestKind(ctx, "download"), input.Config, spec) + download := func(ctx context.Context) ([]byte, string, error) { + return executeProtocolBinaryRequest(withProviderRequestKind(ctx, "download"), input.Config, spec) + } + var data []byte + var mimeType string + if input.Mode == "video" { + data, mimeType, err = runVideoDownload(ctx, taskID, pollPolicy, download) + } else { + data, mimeType, err = download(ctx) + } if err != nil { return nil, fmt.Errorf("声明式协议结果下载失败：%w", err) } @@ -886,7 +848,7 @@ func finishProtocolAdapterResult(ctx context.Context, input canvasGenerationInpu default: return nil, fmt.Errorf("声明式协议结果下载不支持生成模式 %s", input.Mode) } - return finishProtocolResult(ctx, input.Config, input.Mode, downloaded) + return finishProtocolResult(ctx, input.Config, input.Mode, taskID, downloaded, pollPolicy) } func protocolResultHasOutput(mode string, result *protocol.Result) bool { @@ -908,21 +870,13 @@ func protocolResultHasOutput(mode string, result *protocol.Result) bool { } func protocolMediaBytes(ctx context.Context, config providerConfig, reference protocol.MediaReference) ([]byte, string, error) { - if strings.TrimSpace(reference.DataURL) != "" { - mimeType, data, err := decodeProviderDataURL(reference.DataURL) - return data, mimeType, err - } - value := strings.TrimSpace(reference.URL) - if value == "" { - return nil, "", errors.New("声明式协议媒体结果地址为空") - } var data []byte var mimeType string var err error for attempt := 0; attempt < 3; attempt++ { - data, mimeType, err = getProviderExternalBinary(withProviderRequestKind(ctx, "download"), config, value) + data, mimeType, err = protocolMediaBytesOnce(ctx, config, reference) if err == nil { - return data, normalizedMediaMimeType(mimeType, data), nil + return data, mimeType, nil } if attempt == 2 || !retryableProtocolMediaDownload(err) { break @@ -934,6 +888,19 @@ func protocolMediaBytes(ctx context.Context, config providerConfig, reference pr return nil, "", fmt.Errorf("声明式协议媒体结果下载失败：%w", err) } +func protocolMediaBytesOnce(ctx context.Context, config providerConfig, reference protocol.MediaReference) ([]byte, string, error) { + if strings.TrimSpace(reference.DataURL) != "" { + mimeType, data, err := decodeProviderDataURL(reference.DataURL) + return data, mimeType, err + } + value := strings.TrimSpace(reference.URL) + if value == "" { + return nil, "", errors.New("声明式协议媒体结果地址为空") + } + data, mimeType, err := getProviderExternalBinary(withProviderRequestKind(ctx, "download"), config, value) + return data, normalizedMediaMimeType(mimeType, data), err +} + func retryableProtocolMediaDownload(err error) bool { if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { return false diff --git a/backend/internal/app/provider_test.go b/backend/internal/app/provider_test.go index e795a49b4..e8abc76c0 100644 --- a/backend/internal/app/provider_test.go +++ b/backend/internal/app/provider_test.go @@ -213,7 +213,7 @@ func TestRunVideoTaskUsesDeclarativeAgnesJSONProtocol(t *testing.T) { defer server.Close() ctx := withProtocolRegistry(context.Background(), center.registrySnapshot()) - result, err := runVideoTask(ctx, canvasGenerationInput{ + result, err := runVideoTaskForTest(ctx, canvasGenerationInput{ Mode: "video", Prompt: "make it move", Config: providerConfig{BaseURL: server.URL + "/v1", APIKey: "test-key", InterfaceType: "agnes-video", Model: "agnes-video-2.5", VideoSeconds: "5", Size: "16:9", VQuality: "720P"}, @@ -261,7 +261,7 @@ func TestRunVideoTaskDownloadsAuthenticatedDeclarativeResult(t *testing.T) { defer server.Close() ctx := withProtocolRegistry(context.Background(), center.registrySnapshot()) - result, err := runVideoTask(ctx, canvasGenerationInput{ + result, err := runVideoTaskForTest(ctx, canvasGenerationInput{ Mode: "video", Prompt: "cinematic shot", Config: providerConfig{BaseURL: server.URL, APIKey: "test-key", InterfaceType: "newapi", Model: "sora-2", VideoSeconds: "5", Size: "1280x720"}, }) @@ -932,7 +932,7 @@ func TestProviderPayloadErrorCategoryPrefersProviderCodeOverModerationWording(t func TestRunVideoTaskRequiresOfficialPluginWhenRegistryEmpty(t *testing.T) { ctx := withProtocolRegistry(context.Background(), emptyProtocolRegistry) - _, err := runVideoTask(ctx, canvasGenerationInput{ + _, err := runVideoTaskForTest(ctx, canvasGenerationInput{ Prompt: "make it move", Config: providerConfig{BaseURL: "https://example.com", APIKey: "key", Model: "MiniMax-H3", InterfaceType: "minimax-video"}, }) @@ -1629,7 +1629,7 @@ func TestRunVideoTaskUsesNewAPIForAnyVideoModel(t *testing.T) { })) defer server.Close() - result, err := runVideoTask(context.Background(), canvasGenerationInput{ + result, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Prompt: "make it move", Config: providerConfig{BaseURL: server.URL + "/v1", APIKey: "test-key", Model: "custom-video-v1"}, }) @@ -1646,6 +1646,85 @@ func TestRunVideoTaskUsesNewAPIForAnyVideoModel(t *testing.T) { } } +func TestRunVideoTaskRetriesTransientPollFailureWithoutRecreating(t *testing.T) { + t.Setenv("CANVAS_ALLOW_PRIVATE_UPSTREAMS", "true") + createCalls := 0 + pollCalls := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.Method + " " + r.URL.Path { + case "POST /v1/videos": + createCalls++ + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"video-1","status":"queued"}`)) + case "GET /v1/videos/video-1": + pollCalls++ + w.Header().Set("Content-Type", "application/json") + if pollCalls == 1 { + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = w.Write([]byte(`{"message":"temporary outage"}`)) + return + } + _, _ = w.Write([]byte(`{"id":"video-1","status":"completed"}`)) + case "GET /v1/videos/video-1/content": + w.Header().Set("Content-Type", "video/mp4") + _, _ = w.Write([]byte("video")) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + result, err := runVideoTaskWithPolicy(context.Background(), canvasGenerationInput{ + Prompt: "make it move", + Config: providerConfig{BaseURL: server.URL + "/v1", APIKey: "test-key", Model: "custom-video-v1"}, + }, fastVideoPollPolicy()) + if err != nil { + t.Fatal(err) + } + if result["mode"] != "video" || createCalls != 1 || pollCalls != 2 { + t.Fatalf("result = %#v, create calls = %d, poll calls = %d", result, createCalls, pollCalls) + } +} + +func TestRunVideoTaskRetriesDownloadWithoutRepolling(t *testing.T) { + t.Setenv("CANVAS_ALLOW_PRIVATE_UPSTREAMS", "true") + pollCalls := 0 + downloadCalls := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.Method + " " + r.URL.Path { + case "POST /v1/videos": + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"video-1","status":"queued"}`)) + case "GET /v1/videos/video-1": + pollCalls++ + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"video-1","status":"completed"}`)) + case "GET /v1/videos/video-1/content": + downloadCalls++ + if downloadCalls == 1 { + w.WriteHeader(http.StatusBadGateway) + return + } + w.Header().Set("Content-Type", "video/mp4") + _, _ = w.Write([]byte("video")) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + result, err := runVideoTaskWithPolicy(context.Background(), canvasGenerationInput{ + Prompt: "make it move", + Config: providerConfig{BaseURL: server.URL + "/v1", APIKey: "test-key", Model: "custom-video-v1"}, + }, fastVideoPollPolicy()) + if err != nil { + t.Fatal(err) + } + if result["mode"] != "video" || pollCalls != 1 || downloadCalls != 2 { + t.Fatalf("result = %#v, poll calls = %d, download calls = %d", result, pollCalls, downloadCalls) + } +} + func TestRunVideoTaskSendsOnlyDeclaredResolutionName(t *testing.T) { tests := []struct { name string @@ -1689,7 +1768,7 @@ func TestRunVideoTaskSendsOnlyDeclaredResolutionName(t *testing.T) { if test.withoutProfile { profile = nil } - _, err := runVideoTask(context.Background(), canvasGenerationInput{ + _, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Prompt: "synthetic prompt", Config: providerConfig{BaseURL: server.URL + "/v1", APIKey: "test-key", Model: "public-video", VideoSeconds: "8", Size: "16:9", VQuality: test.quality}, VideoCapability: profile, @@ -1731,7 +1810,7 @@ func TestRunVideoTaskUsesNestedURLBeforeResultURL(t *testing.T) { })) defer server.Close() - result, err := runVideoTask(context.Background(), canvasGenerationInput{ + result, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Prompt: "make it move", Config: providerConfig{BaseURL: server.URL, APIKey: "test-key", Model: "grok-imagine-video-1.5-1080p", VideoSeconds: "15"}, }) @@ -1782,7 +1861,7 @@ func TestRunVideoTaskUsesJSONForGrokVideo(t *testing.T) { })) defer server.Close() - result, err := runVideoTask(context.Background(), canvasGenerationInput{ + result, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Prompt: "make it move", Config: providerConfig{BaseURL: server.URL + "/v1", APIKey: "test-key", Model: "grok-video", VideoSeconds: "10"}, ReferenceImages: []providerMedia{{ID: "image-1", DataURL: testReferenceImageDataURL}}, @@ -1840,7 +1919,7 @@ func TestRunVideoTaskUsesXAIVideoGenerationEndpoint(t *testing.T) { })) defer server.Close() - result, err := runVideoTask(context.Background(), canvasGenerationInput{ + result, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Prompt: "make it move", Config: providerConfig{ BaseURL: server.URL + "/v1", @@ -1882,7 +1961,7 @@ func TestRunVideoTaskXAIVideoRejectsUnreachableResultURL(t *testing.T) { })) defer server.Close() - _, err := runVideoTask(context.Background(), canvasGenerationInput{ + _, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Prompt: "make it move", Config: providerConfig{ BaseURL: server.URL + "/v1", @@ -2061,7 +2140,7 @@ func TestVolcengineArkVideoProtocolUsesContentTaskAndDownloadsResult(t *testing. })) defer server.Close() - result, err := runVideoTask(context.Background(), canvasGenerationInput{ + result, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Prompt: "make it move", Config: providerConfig{BaseURL: server.URL + "/api/v3", APIKey: "test-key", Model: "doubao-seedance-test", InterfaceType: "volcengine-ark-video"}, ReferenceImages: []providerMedia{{ID: "start", URL: server.URL + "/reference.png"}}, @@ -2257,7 +2336,7 @@ func TestRunNewAPIChannel1VideoTaskDownloadsSucceededObject(t *testing.T) { })) defer server.Close() - result, err := runVideoTask(context.Background(), canvasGenerationInput{ + result, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Prompt: "make it move", Config: providerConfig{BaseURL: server.URL + "/v1", APIKey: "test-key", Model: "seedance-2.0", InterfaceType: "newapi-channel-1"}, }) @@ -2321,7 +2400,7 @@ func TestRunNewAPIChannel2VideoTaskDownloadsTemporaryResult(t *testing.T) { profile := DefaultModelCapabilityConfigForModel("newapi-channel-2", "grok-image-video").Video profile.Resolutions = []string{"720p"} profile.DefaultResolution = "720p" - result, err := runVideoTask(context.Background(), canvasGenerationInput{ + result, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Prompt: "make it move", Config: providerConfig{BaseURL: server.URL, APIKey: "test-key", Model: "grok-image-video", InterfaceType: "newapi-channel-2", VideoSeconds: "15", Size: "9:16", VQuality: "720", VideoGenerateAudio: "true"}, ReferenceImages: []providerMedia{ @@ -2376,7 +2455,7 @@ func TestRunNewAPIChannel2VideoTaskResumesOriginalProviderTaskWithoutAnotherPost ctx := withProviderAnalytics(context.Background(), nil, model.Task{ ID: "task-1", Type: "canvas_video", ProviderRequestID: "existing-provider-task", InputJSON: string(inputJSON), }) - result, err := runVideoTask(ctx, input) + result, err := runVideoTaskForTest(ctx, input) if err != nil { t.Fatalf("runVideoTask() error = %v", err) } @@ -2393,7 +2472,7 @@ func TestRunNewAPIChannel2VideoTaskReturnsTypedDeadlineWhenPollingWindowEnds(t * ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second)) defer cancel() ctx = withProviderAnalytics(ctx, nil, model.Task{ID: "task-1", Type: "canvas_video", ProviderRequestID: "existing-provider-task"}) - _, err := runVideoTask(ctx, canvasGenerationInput{ + _, err := runVideoTaskForTest(ctx, canvasGenerationInput{ Mode: "video", Config: providerConfig{InterfaceType: string(model.ChannelInterfaceNewAPIChannel2), Model: "video-model", BaseURL: "https://example.com", APIKey: "key"}, }) @@ -2427,7 +2506,7 @@ func TestRunGeminiVeoVideoTaskUsesLongRunningOperation(t *testing.T) { })) defer server.Close() - result, err := runVideoTask(context.Background(), canvasGenerationInput{ + result, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Prompt: "make it move", Config: providerConfig{BaseURL: server.URL, APIKey: "test-key", APIFormat: "gemini", Model: "veo-test", InterfaceType: "gemini-veo", VideoSeconds: "6", Size: "16:9", VQuality: "720"}, }) @@ -2632,7 +2711,7 @@ func TestRunNovitaVideoTaskDownloadsSucceededVideo(t *testing.T) { })) defer server.Close() - result, err := runVideoTask(context.Background(), canvasGenerationInput{ + result, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Prompt: "make it move", Config: providerConfig{BaseURL: server.URL, APIKey: "test-key", Model: "kling2.5_turbo_pro_t2v", InterfaceType: "novita-video", VideoSeconds: "5", Size: "16:9"}, }) @@ -2665,7 +2744,7 @@ func TestRunNovitaVideoTaskReturnsFailureReason(t *testing.T) { })) defer server.Close() - _, err := runVideoTask(context.Background(), canvasGenerationInput{ + _, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Prompt: "make it move", Config: providerConfig{BaseURL: server.URL, APIKey: "test-key", Model: "kling2.5_turbo_pro_t2v", InterfaceType: "novita-video"}, }) @@ -2711,7 +2790,7 @@ func TestRunMiniMaxVideoTaskCreatesPollsAndDownloads(t *testing.T) { })) defer server.Close() - result, err := runVideoTask(context.Background(), canvasGenerationInput{ + result, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Mode: "video", Prompt: "make it move", Config: providerConfig{BaseURL: server.URL, APIKey: "test-key", Model: "MiniMax-H3", InterfaceType: "minimax-video", VideoSeconds: "5", VQuality: "768P", Size: "16:9"}, @@ -2747,7 +2826,7 @@ func TestRunMiniMaxVideoTaskSendsOfficial2KResolution(t *testing.T) { _, _ = w.Write([]byte(`{}`)) })) defer server.Close() - _, err := runVideoTask(context.Background(), canvasGenerationInput{ + _, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Mode: "video", Prompt: "make it move", Config: providerConfig{BaseURL: server.URL, APIKey: "test-key", Model: "MiniMax-H3", InterfaceType: "minimax-video", VideoSeconds: "6", VQuality: "2K"}, }) @@ -2792,7 +2871,7 @@ func TestRunMiniMaxVideoTaskUsesExplicitReferenceRoles(t *testing.T) { })) defer server.Close() - _, err := runVideoTask(context.Background(), canvasGenerationInput{ + _, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Mode: "video", Prompt: "保持角色一致", Config: providerConfig{BaseURL: server.URL, APIKey: "test-key", Model: "MiniMax-H3", InterfaceType: "minimax-video", VideoSeconds: "6", VQuality: "768P", Size: "16:9"}, @@ -2821,7 +2900,7 @@ func TestRunMiniMaxVideoTaskReturnsFailureReason(t *testing.T) { })) defer server.Close() - _, err := runVideoTask(context.Background(), canvasGenerationInput{ + _, err := runVideoTaskForTest(context.Background(), canvasGenerationInput{ Mode: "video", Prompt: "make it move", Config: providerConfig{BaseURL: server.URL, APIKey: "test-key", Model: "MiniMax-H3", InterfaceType: "minimax-video", VideoSeconds: "6"}, diff --git a/backend/internal/app/provider_video.go b/backend/internal/app/provider_video.go index 05a4203b4..51884a221 100644 --- a/backend/internal/app/provider_video.go +++ b/backend/internal/app/provider_video.go @@ -10,7 +10,6 @@ import ( "mime/multipart" "strconv" "strings" - "time" "infinite-canvas/backend/internal/model" ) @@ -52,6 +51,10 @@ func (s *Service) validateResolvedVideoCapability(input *canvasGenerationInput) } func runVideoTask(ctx context.Context, input canvasGenerationInput) (map[string]interface{}, error) { + return runVideoTaskWithPolicy(ctx, input, defaultVideoPollPolicy()) +} + +func runVideoTaskWithPolicy(ctx context.Context, input canvasGenerationInput, pollPolicy videoPollPolicy) (map[string]interface{}, error) { // 路由顺序是协议边界，不是“哪个请求先试”：官方声明式接口必须由已注册适配器执行， // 缺少适配器时直接失败，不能偷偷退回遗留手写协议；只有未声明为官方插件的旧渠道才继续走兼容分支。 if strings.TrimSpace(input.Mode) == "" { @@ -60,17 +63,17 @@ func runVideoTask(ctx context.Context, input canvasGenerationInput) (map[string] // 已有官方声明式插件的 InterfaceType 只走适配器。未注入 registry 时补官方包， // 显式空 registry 则报“插件未安装”，不再回退到手写协议。 ctx = ensureOfficialProtocolAdapter(ctx, input.Config.InterfaceType) - if _, ok := declarativeProtocolAdapterForContext(ctx, input.Config.InterfaceType); ok { - return runDeclarativeProtocolTask(ctx, input) + if adapter, ok := declarativeProtocolAdapterForContext(ctx, input.Config.InterfaceType); ok { + return runProtocolAdapterTaskWithPolicy(ctx, input, adapter, pollPolicy) } if label, official := officialDeclarativeVideoInterface(input.Config.InterfaceType); official { return nil, fmt.Errorf("%s 视频插件未安装", label) } if isArkPlanVideoConfig(input.Config) { - return runSeedanceAgentPlanVideoTask(ctx, input) + return runSeedanceAgentPlanVideoTask(ctx, input, pollPolicy) } if isSeedanceVideoConfig(input.Config) { - return runSeedanceVideosTask(ctx, input) + return runSeedanceVideosTask(ctx, input, pollPolicy) } if len(input.ReferenceVideos) > 0 || len(input.ReferenceAudios) > 0 { return nil, errors.New("OpenAI 风格视频接口不支持参考视频或参考音频，请切换到 Seedance / Agent Plan 渠道") @@ -131,10 +134,10 @@ func runVideoTask(ctx context.Context, input canvasGenerationInput) (map[string] if id == "" { return nil, errors.New("视频接口没有返回任务 ID") } - for deadline := providerPollingDeadline(ctx); time.Now().Before(deadline); { + return runVideoPollLoop(ctx, id, pollPolicy, func(ctx context.Context) (videoPollOutcome, error) { var state map[string]interface{} if err := getJSON(ctx, input.Config, "/videos/"+id, &state); err != nil { - return nil, err + return videoPollOutcome{}, err } if data, ok := state["data"].(map[string]interface{}); ok { state = data @@ -142,27 +145,28 @@ func runVideoTask(ctx context.Context, input canvasGenerationInput) (map[string] status := strings.ToLower(stringField(state, "status")) if status == "completed" || status == "succeeded" || status == "success" || status == "done" { if videoURL := newAPIVideoResultURL(state); videoURL != "" { - data, mimeType, err := getProviderExternalBinary(withProviderRequestKind(ctx, "download"), input.Config, videoURL) + data, mimeType, err := runVideoDownload(ctx, id, pollPolicy, func(ctx context.Context) ([]byte, string, error) { + return getProviderExternalBinary(withProviderRequestKind(ctx, "download"), input.Config, videoURL) + }) if err != nil { - return nil, fmt.Errorf("视频结果下载失败（任务 %s）：%w", id, err) + return videoPollOutcome{}, fmt.Errorf("视频结果下载失败（任务 %s）：%w", id, err) } mimeType = normalizedMediaMimeType(mimeType, data) - return map[string]interface{}{"mode": "video", "video": map[string]interface{}{"dataUrl": dataURL(mimeType, data), "mimeType": mimeType}}, nil + return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video", "video": map[string]interface{}{"dataUrl": dataURL(mimeType, data), "mimeType": mimeType}}}, nil } - data, mimeType, err := getBinary(ctx, input.Config, "/videos/"+id+"/content") + data, mimeType, err := runVideoDownload(ctx, id, pollPolicy, func(ctx context.Context) ([]byte, string, error) { + return getBinary(withProviderRequestKind(ctx, "download"), input.Config, "/videos/"+id+"/content") + }) if err != nil { - return nil, err + return videoPollOutcome{}, err } - return map[string]interface{}{"mode": "video", "video": map[string]interface{}{"dataUrl": dataURL(mimeType, data), "mimeType": mimeType}}, nil + return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video", "video": map[string]interface{}{"dataUrl": dataURL(mimeType, data), "mimeType": mimeType}}}, nil } if status == "failed" || status == "cancelled" { - return nil, errors.New("视频生成失败") + return videoPollOutcome{}, errors.New("视频生成失败") } - if err := sleepContext(ctx, 2500*time.Millisecond); err != nil { - return nil, err - } - } - return nil, errors.New("视频生成超时") + return videoPollOutcome{}, nil + }) } func newAPIVideoResultURL(state map[string]interface{}) string { @@ -221,7 +225,7 @@ func grokVideoBody(input canvasGenerationInput) (map[string]interface{}, error) return body, nil } -func runSeedanceVideosTask(ctx context.Context, input canvasGenerationInput) (map[string]interface{}, error) { +func runSeedanceVideosTask(ctx context.Context, input canvasGenerationInput, pollPolicy videoPollPolicy) (map[string]interface{}, error) { // 恢复任务已有 provider ID 时只能继续查询，绝不能重新 create，否则会产生第二个计费任务。 // create 成功后的 poll/download 任一失败都保留失败或结果未知语义，不降级成“成功但无内容”。 id := resumedProviderRequestID(ctx) @@ -246,10 +250,10 @@ func runSeedanceVideosTask(ctx context.Context, input canvasGenerationInput) (ma if id == "" { return nil, errors.New("Seedance 接口没有返回任务 ID") } - for deadline := providerPollingDeadline(ctx); time.Now().Before(deadline); { + return runVideoPollLoop(ctx, id, pollPolicy, func(ctx context.Context) (videoPollOutcome, error) { var state map[string]interface{} if err := getJSON(ctx, input.Config, "/videos/"+id, &state); err != nil { - return nil, err + return videoPollOutcome{}, err } if data, ok := state["data"].(map[string]interface{}); ok { state = data @@ -258,29 +262,30 @@ func runSeedanceVideosTask(ctx context.Context, input canvasGenerationInput) (ma if status == "completed" || status == "succeeded" { videoURL := stringField(state, "video_url") if videoURL != "" { - data, mimeType, err := getExternalBinary(withProviderRequestKind(ctx, "download"), videoURL) + data, mimeType, err := runVideoDownload(ctx, id, pollPolicy, func(ctx context.Context) ([]byte, string, error) { + return getExternalBinary(withProviderRequestKind(ctx, "download"), videoURL) + }) if err != nil { - return nil, fmt.Errorf("视频结果下载失败：%w", err) + return videoPollOutcome{}, fmt.Errorf("视频结果下载失败：%w", err) } - return map[string]interface{}{"mode": "video", "video": map[string]interface{}{"dataUrl": dataURL(mimeType, data), "mimeType": mimeType}}, nil + return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video", "video": map[string]interface{}{"dataUrl": dataURL(mimeType, data), "mimeType": mimeType}}}, nil } - data, mimeType, err := getBinary(ctx, input.Config, "/videos/"+id+"/content") + data, mimeType, err := runVideoDownload(ctx, id, pollPolicy, func(ctx context.Context) ([]byte, string, error) { + return getBinary(withProviderRequestKind(ctx, "download"), input.Config, "/videos/"+id+"/content") + }) if err != nil { - return nil, fmt.Errorf("Seedance 任务成功但未返回视频 URL，备用内容下载失败：%w", err) + return videoPollOutcome{}, fmt.Errorf("Seedance 任务成功但未返回视频 URL，备用内容下载失败：%w", err) } - return map[string]interface{}{"mode": "video", "video": map[string]interface{}{"dataUrl": dataURL(mimeType, data), "mimeType": mimeType}}, nil + return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video", "video": map[string]interface{}{"dataUrl": dataURL(mimeType, data), "mimeType": mimeType}}}, nil } if status == "failed" || status == "cancelled" || status == "expired" { - return nil, errors.New(defaultString(seedanceErrorMessage(state), "Seedance 视频生成失败")) + return videoPollOutcome{}, errors.New(defaultString(seedanceErrorMessage(state), "Seedance 视频生成失败")) } - if err := sleepContext(ctx, 5*time.Second); err != nil { - return nil, err - } - } - return nil, errors.New("Seedance 视频生成超时") + return videoPollOutcome{}, nil + }) } -func runSeedanceAgentPlanVideoTask(ctx context.Context, input canvasGenerationInput) (map[string]interface{}, error) { +func runSeedanceAgentPlanVideoTask(ctx context.Context, input canvasGenerationInput, pollPolicy videoPollPolicy) (map[string]interface{}, error) { providerName := "Seedance" if input.Config.InterfaceType == string(model.ChannelInterfaceVolcengineArkVideo) { providerName = "火山方舟" @@ -331,10 +336,10 @@ func runSeedanceAgentPlanVideoTask(ctx context.Context, input canvasGenerationIn if id == "" { return nil, fmt.Errorf("%s接口没有返回任务 ID", providerName) } - for deadline := providerPollingDeadline(ctx); time.Now().Before(deadline); { + return runVideoPollLoop(ctx, id, pollPolicy, func(ctx context.Context) (videoPollOutcome, error) { var state map[string]interface{} if err := getJSON(ctx, input.Config, "/contents/generations/tasks/"+id, &state); err != nil { - return nil, err + return videoPollOutcome{}, err } if data, ok := state["data"].(map[string]interface{}); ok { state = data @@ -344,22 +349,21 @@ func runSeedanceAgentPlanVideoTask(ctx context.Context, input canvasGenerationIn content, _ := state["content"].(map[string]interface{}) videoURL := stringField(content, "video_url") if videoURL == "" { - return nil, fmt.Errorf("%s任务成功但没有返回视频 URL", providerName) + return videoPollOutcome{}, fmt.Errorf("%s任务成功但没有返回视频 URL", providerName) } - data, mimeType, err := getExternalBinary(withProviderRequestKind(ctx, "download"), videoURL) + data, mimeType, err := runVideoDownload(ctx, id, pollPolicy, func(ctx context.Context) ([]byte, string, error) { + return getExternalBinary(withProviderRequestKind(ctx, "download"), videoURL) + }) if err != nil { - return nil, fmt.Errorf("视频结果下载失败：%w", err) + return videoPollOutcome{}, fmt.Errorf("视频结果下载失败：%w", err) } - return map[string]interface{}{"mode": "video", "video": map[string]interface{}{"dataUrl": dataURL(mimeType, data), "mimeType": mimeType}}, nil + return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video", "video": map[string]interface{}{"dataUrl": dataURL(mimeType, data), "mimeType": mimeType}}}, nil } if status == "failed" || status == "cancelled" || status == "expired" { - return nil, fmt.Errorf("%s视频生成失败", providerName) + return videoPollOutcome{}, fmt.Errorf("%s视频生成失败", providerName) } - if err := sleepContext(ctx, 5*time.Second); err != nil { - return nil, err - } - } - return nil, fmt.Errorf("%s视频生成超时", providerName) + return videoPollOutcome{}, nil + }) } func seedanceContent(input canvasGenerationInput) ([]map[string]interface{}, error) { diff --git a/backend/internal/app/provider_video_polling.go b/backend/internal/app/provider_video_polling.go new file mode 100644 index 000000000..ce404f619 --- /dev/null +++ b/backend/internal/app/provider_video_polling.go @@ -0,0 +1,273 @@ +package app + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "strings" + "time" +) + +const defaultVideoPollInterval = 30 * time.Second + +type videoPollPolicy struct { + InitialDelay time.Duration + Interval time.Duration + MaxNotFoundMisses int + MaxMalformedResponses int + MaxDownloadTries int + RetryTransient bool + Sleep func(context.Context, time.Duration) error + Notify func(context.Context, string, videoPollEvent, error) +} + +type videoPollEvent string + +const ( + videoPollEventRetrying videoPollEvent = "retrying" + videoPollEventRecovered videoPollEvent = "recovered" +) + +type videoPollOutcome struct { + Done bool + Result map[string]interface{} +} + +type videoDownloadError struct { + TaskID string + Cause error +} + +func (e videoDownloadError) Error() string { + if strings.TrimSpace(e.TaskID) == "" { + return fmt.Sprintf("视频结果下载失败：%v", e.Cause) + } + return fmt.Sprintf("视频结果下载失败（任务 %s）：%v", e.TaskID, e.Cause) +} + +func (e videoDownloadError) Unwrap() error { return e.Cause } + +func defaultVideoPollPolicy() videoPollPolicy { + return videoPollPolicy{ + InitialDelay: defaultVideoPollInterval, + Interval: defaultVideoPollInterval, + MaxNotFoundMisses: 3, + MaxMalformedResponses: 3, + MaxDownloadTries: 3, + RetryTransient: true, + Sleep: sleepContext, + Notify: notifyTaskVideoPollEvent, + } +} + +func runVideoPollLoop(ctx context.Context, taskID string, policy videoPollPolicy, query func(context.Context) (videoPollOutcome, error)) (map[string]interface{}, error) { + policy = normalizeVideoPollPolicy(policy) + deadline := providerPollingDeadline(ctx) + pollContext, cancel := context.WithDeadline(ctx, deadline) + defer cancel() + ctx = pollContext + nextDelay := policy.InitialDelay + notFoundMisses := 0 + malformedResponses := 0 + retrying := false + for time.Now().Before(deadline) { + if nextDelay > 0 { + if err := policy.Sleep(ctx, nextDelay); err != nil { + return nil, err + } + } + outcome, err := query(ctx) + if err != nil { + if !policy.RetryTransient { + return nil, err + } + retry, notFound := retryableVideoPollError(ctx, err) + if !retry { + return nil, err + } + malformed := isTransientResponseDecodeError(err) + if malformed { + malformedResponses++ + notFoundMisses = 0 + if malformedResponses >= policy.MaxMalformedResponses { + return nil, err + } + } else if notFound { + malformedResponses = 0 + notFoundMisses++ + if notFoundMisses >= policy.MaxNotFoundMisses { + return nil, err + } + } else { + malformedResponses = 0 + notFoundMisses = 0 + } + if !retrying { + retrying = true + policy.Notify(ctx, taskID, videoPollEventRetrying, err) + } + nextDelay = max(policy.Interval, providerRetryAfter(err)) + continue + } + notFoundMisses = 0 + malformedResponses = 0 + nextDelay = policy.Interval + if retrying { + retrying = false + policy.Notify(ctx, taskID, videoPollEventRecovered, nil) + } + if outcome.Done { + return outcome.Result, nil + } + } + if err := ctx.Err(); err != nil { + return nil, err + } + return nil, context.DeadlineExceeded +} + +func normalizeVideoPollPolicy(policy videoPollPolicy) videoPollPolicy { + if policy.Interval <= 0 { + policy.Interval = defaultVideoPollInterval + } + if policy.MaxNotFoundMisses <= 0 { + policy.MaxNotFoundMisses = 3 + } + if policy.MaxMalformedResponses <= 0 { + policy.MaxMalformedResponses = 3 + } + if policy.MaxDownloadTries <= 0 { + policy.MaxDownloadTries = 3 + } + if policy.Sleep == nil { + policy.Sleep = sleepContext + } + if policy.Notify == nil { + policy.Notify = func(context.Context, string, videoPollEvent, error) {} + } + return policy +} + +func runVideoDownload(ctx context.Context, taskID string, policy videoPollPolicy, download func(context.Context) ([]byte, string, error)) ([]byte, string, error) { + policy = normalizeVideoPollPolicy(policy) + var lastErr error + for attempt := 1; attempt <= policy.MaxDownloadTries; attempt++ { + data, mimeType, err := download(ctx) + if err == nil { + return data, mimeType, nil + } + lastErr = err + retry, _ := retryableVideoPollError(ctx, err) + if !retry || attempt == policy.MaxDownloadTries { + return nil, "", videoDownloadError{TaskID: taskID, Cause: err} + } + delay := max(policy.Interval, providerRetryAfter(err)) + if err := policy.Sleep(ctx, delay); err != nil { + return nil, "", err + } + } + return nil, "", videoDownloadError{TaskID: taskID, Cause: lastErr} +} + +func retryableVideoPollError(ctx context.Context, err error) (retry bool, notFound bool) { + if err == nil || ctx.Err() != nil || errors.Is(err, context.Canceled) { + return false, false + } + var downloadError videoDownloadError + if errors.As(err, &downloadError) { + return false, false + } + if code, _ := ChannelSlotFailureDetails(err); code != "" { + return true, false + } + var circuitOpen providerCircuitOpenError + if errors.As(err, &circuitOpen) { + return true, false + } + if isTransientResponseDecodeError(err) { + return true, false + } + if errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, io.ErrClosedPipe) { + return true, false + } + if errors.Is(err, context.DeadlineExceeded) { + return true, false + } + var httpErr providerHTTPError + if errors.As(err, &httpErr) { + if isProviderTaskNotReadyError(httpErr) { + return true, true + } + if httpErr.StatusCode == http.StatusNotFound { + return true, true + } + switch httpErr.StatusCode { + case http.StatusRequestTimeout, http.StatusConflict, http.StatusTooEarly, http.StatusTooManyRequests: + return true, false + default: + return httpErr.StatusCode >= http.StatusInternalServerError, false + } + } + var networkError net.Error + if errors.As(err, &networkError) { + return networkError.Timeout() || networkError.Temporary(), false + } + return false, false +} + +func isTransientResponseDecodeError(err error) bool { + var decodeError providerResponseDecodeError + if errors.As(err, &decodeError) { + return true + } + var syntaxError *json.SyntaxError + return errors.As(err, &syntaxError) +} + +func notifyTaskVideoPollEvent(ctx context.Context, _ string, event videoPollEvent, eventErr error) { + metadata, ok := ctx.Value(providerAnalyticsKey{}).(providerAnalyticsContext) + if !ok || metadata.Service == nil || metadata.Service.repo == nil || metadata.TaskID == "" { + return + } + message := "上游视频查询暂时异常，将继续轮询原任务" + level := "warn" + payload := "" + if eventErr != nil { + payload = eventErr.Error() + } + if event == videoPollEventRecovered { + message = "上游视频查询已恢复" + level = "info" + } + _ = metadata.Service.log(metadata.UserID, metadata.TaskID, level, message, payload) +} + +func isProviderTaskNotReadyError(httpErr providerHTTPError) bool { + if httpErr.StatusCode != http.StatusBadRequest && httpErr.StatusCode != http.StatusNotFound { + return false + } + var payload map[string]any + if json.Unmarshal([]byte(httpErr.Body), &payload) != nil { + return false + } + code, message := providerFailureDetails(payload) + for _, value := range []string{code, message} { + switch strings.ToLower(strings.TrimSpace(value)) { + case "task_not_exist", "task_not_found", "task not exist", "task not found": + return true + } + } + return false +} + +func providerRetryAfter(err error) time.Duration { + var httpErr providerHTTPError + if errors.As(err, &httpErr) && httpErr.RetryAfter > 0 { + return httpErr.RetryAfter + } + return 0 +} diff --git a/backend/internal/app/provider_video_polling_test.go b/backend/internal/app/provider_video_polling_test.go new file mode 100644 index 000000000..00c8c8604 --- /dev/null +++ b/backend/internal/app/provider_video_polling_test.go @@ -0,0 +1,266 @@ +package app + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "reflect" + "testing" + "time" +) + +func TestVideoPollRecoversFromTransientHTTPError(t *testing.T) { + attempts := 0 + result, err := runVideoPollLoop(context.Background(), "provider-task-1", fastVideoPollPolicy(), func(context.Context) (videoPollOutcome, error) { + attempts++ + switch attempts { + case 1: + return videoPollOutcome{}, providerHTTPError{StatusCode: http.StatusBadGateway, Status: "502 Bad Gateway"} + case 2: + return videoPollOutcome{}, nil + default: + return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video"}}, nil + } + }) + if err != nil { + t.Fatal(err) + } + if attempts != 3 || result["mode"] != "video" { + t.Fatalf("attempts = %d, result = %#v", attempts, result) + } +} + +func TestVideoPollWaitsBeforeFirstQueryAndUsesRetryAfter(t *testing.T) { + var waits []time.Duration + policy := fastVideoPollPolicy() + policy.InitialDelay = 10 * time.Millisecond + policy.Interval = 10 * time.Millisecond + policy.Sleep = func(_ context.Context, delay time.Duration) error { + waits = append(waits, delay) + return nil + } + attempts := 0 + _, err := runVideoPollLoop(context.Background(), "provider-task-1", policy, func(context.Context) (videoPollOutcome, error) { + attempts++ + if attempts == 1 { + return videoPollOutcome{}, providerHTTPError{StatusCode: http.StatusTooManyRequests, RetryAfter: 20 * time.Millisecond} + } + return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video"}}, nil + }) + if err != nil { + t.Fatal(err) + } + if want := []time.Duration{10 * time.Millisecond, 20 * time.Millisecond}; !reflect.DeepEqual(waits, want) { + t.Fatalf("waits = %v, want %v", waits, want) + } +} + +func TestVideoPollStopsImmediatelyOnAuthenticationError(t *testing.T) { + attempts := 0 + _, err := runVideoPollLoop(context.Background(), "provider-task-1", fastVideoPollPolicy(), func(context.Context) (videoPollOutcome, error) { + attempts++ + return videoPollOutcome{}, providerHTTPError{StatusCode: http.StatusUnauthorized, Status: "401 Unauthorized"} + }) + var httpErr providerHTTPError + if !errors.As(err, &httpErr) || httpErr.StatusCode != http.StatusUnauthorized { + t.Fatalf("error = %#v, want 401 provider error", err) + } + if attempts != 1 { + t.Fatalf("attempts = %d, want 1", attempts) + } +} + +func TestVideoPollStopsAfterThreeConsecutiveNotFoundResponses(t *testing.T) { + attempts := 0 + _, err := runVideoPollLoop(context.Background(), "provider-task-1", fastVideoPollPolicy(), func(context.Context) (videoPollOutcome, error) { + attempts++ + return videoPollOutcome{}, providerHTTPError{StatusCode: http.StatusNotFound, Status: "404 Not Found"} + }) + var httpErr providerHTTPError + if !errors.As(err, &httpErr) || httpErr.StatusCode != http.StatusNotFound { + t.Fatalf("error = %#v, want 404 provider error", err) + } + if attempts != 3 { + t.Fatalf("attempts = %d, want 3", attempts) + } +} + +func TestVideoPollSuccessfulPendingResponseResetsNotFoundCounter(t *testing.T) { + attempts := 0 + result, err := runVideoPollLoop(context.Background(), "provider-task-1", fastVideoPollPolicy(), func(context.Context) (videoPollOutcome, error) { + attempts++ + switch attempts { + case 1, 3: + return videoPollOutcome{}, providerHTTPError{StatusCode: http.StatusNotFound, Status: "404 Not Found"} + case 2: + return videoPollOutcome{}, nil + default: + return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video"}}, nil + } + }) + if err != nil { + t.Fatal(err) + } + if attempts != 4 || result["mode"] != "video" { + t.Fatalf("attempts = %d, result = %#v", attempts, result) + } +} + +func TestVideoPollRetriesMalformedResponseTwiceThenSucceeds(t *testing.T) { + attempts := 0 + result, err := runVideoPollLoop(context.Background(), "provider-task-1", fastVideoPollPolicy(), func(context.Context) (videoPollOutcome, error) { + attempts++ + if attempts <= 2 { + return videoPollOutcome{}, &json.SyntaxError{Offset: int64(attempts)} + } + return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video"}}, nil + }) + if err != nil || attempts != 3 || result["mode"] != "video" { + t.Fatalf("result = %#v, error = %v, attempts = %d", result, err, attempts) + } +} + +func TestVideoPollStopsAfterThreeMalformedResponses(t *testing.T) { + attempts := 0 + _, err := runVideoPollLoop(context.Background(), "provider-task-1", fastVideoPollPolicy(), func(context.Context) (videoPollOutcome, error) { + attempts++ + return videoPollOutcome{}, &json.SyntaxError{Offset: int64(attempts)} + }) + var syntaxError *json.SyntaxError + if !errors.As(err, &syntaxError) || attempts != 3 { + t.Fatalf("error = %#v, attempts = %d, want three attempts and syntax error", err, attempts) + } +} + +func TestVideoPollTreatsCircuitAndInterruptedReadsAsTransient(t *testing.T) { + for _, err := range []error{providerCircuitOpenError{}, io.ErrUnexpectedEOF, io.ErrClosedPipe} { + retry, notFound := retryableVideoPollError(context.Background(), err) + if !retry || notFound { + t.Fatalf("error %T classified as retry=%v notFound=%v", err, retry, notFound) + } + } +} + +func TestVideoPollNotifiesOnlyWhenRetryStartsAndRecovers(t *testing.T) { + var events []videoPollEvent + policy := fastVideoPollPolicy() + policy.Notify = func(_ context.Context, _ string, event videoPollEvent, _ error) { + events = append(events, event) + } + attempts := 0 + _, err := runVideoPollLoop(context.Background(), "provider-task-1", policy, func(context.Context) (videoPollOutcome, error) { + attempts++ + switch attempts { + case 1, 2: + return videoPollOutcome{}, providerHTTPError{StatusCode: http.StatusBadGateway} + case 3: + return videoPollOutcome{}, nil + default: + return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video"}}, nil + } + }) + if err != nil { + t.Fatal(err) + } + want := []videoPollEvent{videoPollEventRetrying, videoPollEventRecovered} + if len(events) != len(want) || events[0] != want[0] || events[1] != want[1] { + t.Fatalf("events = %#v, want %#v", events, want) + } +} + +func TestVideoPollContextCancellationInterruptsInitialWait(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + called := false + _, err := runVideoPollLoop(ctx, "provider-task-1", defaultVideoPollPolicy(), func(context.Context) (videoPollOutcome, error) { + called = true + return videoPollOutcome{}, nil + }) + if !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v, want context canceled", err) + } + if called { + t.Fatal("query ran after cancellation") + } +} + +func TestVideoPollAppliesFallbackDeadlineToWaitAndQuery(t *testing.T) { + policy := fastVideoPollPolicy() + waitHadDeadline := false + queryHadDeadline := false + policy.Sleep = func(ctx context.Context, _ time.Duration) error { + _, waitHadDeadline = ctx.Deadline() + return nil + } + _, err := runVideoPollLoop(context.Background(), "provider-task-1", policy, func(ctx context.Context) (videoPollOutcome, error) { + _, queryHadDeadline = ctx.Deadline() + return videoPollOutcome{Done: true, Result: map[string]interface{}{"mode": "video"}}, nil + }) + if err != nil { + t.Fatal(err) + } + if !waitHadDeadline || !queryHadDeadline { + t.Fatalf("wait deadline = %v, query deadline = %v", waitHadDeadline, queryHadDeadline) + } +} + +func TestVideoDownloadRetriesTransientFailure(t *testing.T) { + attempts := 0 + waits := 0 + policy := fastVideoPollPolicy() + policy.Sleep = func(context.Context, time.Duration) error { + waits++ + return nil + } + data, mimeType, err := runVideoDownload(context.Background(), "provider-task-1", policy, func(context.Context) ([]byte, string, error) { + attempts++ + if attempts == 1 { + return nil, "", providerHTTPError{StatusCode: http.StatusBadGateway, Status: "502 Bad Gateway"} + } + return []byte("video"), "video/mp4", nil + }) + if err != nil { + t.Fatal(err) + } + if string(data) != "video" || mimeType != "video/mp4" || attempts != 2 || waits != 1 { + t.Fatalf("data = %q, mime = %q, attempts = %d, waits = %d", data, mimeType, attempts, waits) + } +} + +func TestVideoDownloadStopsAfterThreeTransientFailures(t *testing.T) { + attempts := 0 + _, _, err := runVideoDownload(context.Background(), "provider-task-1", fastVideoPollPolicy(), func(context.Context) ([]byte, string, error) { + attempts++ + return nil, "", providerHTTPError{StatusCode: http.StatusServiceUnavailable, Status: "503 Service Unavailable"} + }) + var httpErr providerHTTPError + if !errors.As(err, &httpErr) || attempts != 3 { + t.Fatalf("error = %#v, attempts = %d, want three attempts and provider error", err, attempts) + } +} + +func TestVideoPollDoesNotRetryExhaustedDownload(t *testing.T) { + attempts := 0 + _, err := runVideoPollLoop(context.Background(), "provider-task-1", fastVideoPollPolicy(), func(context.Context) (videoPollOutcome, error) { + attempts++ + return videoPollOutcome{}, videoDownloadError{TaskID: "provider-task-1", Cause: providerHTTPError{StatusCode: http.StatusBadGateway}} + }) + var downloadError videoDownloadError + if !errors.As(err, &downloadError) || attempts != 1 { + t.Fatalf("error = %#v, poll attempts = %d, want one exhausted download", err, attempts) + } +} + +func fastVideoPollPolicy() videoPollPolicy { + policy := defaultVideoPollPolicy() + policy.InitialDelay = time.Millisecond + policy.Interval = time.Millisecond + policy.Sleep = func(context.Context, time.Duration) error { return nil } + return policy +} + +func runVideoTaskForTest(ctx context.Context, input canvasGenerationInput) (map[string]interface{}, error) { + return runVideoTaskWithPolicy(ctx, input, fastVideoPollPolicy()) +} diff --git a/backend/internal/app/runtime_policy_test.go b/backend/internal/app/runtime_policy_test.go index 026b3c1db..2b63c2fc2 100644 --- a/backend/internal/app/runtime_policy_test.go +++ b/backend/internal/app/runtime_policy_test.go @@ -26,6 +26,25 @@ func TestVideoTaskTimeoutHasFiveMinuteSafetyFloor(t *testing.T) { } } +func TestDefaultVideoTaskTimeoutIsOneHour(t *testing.T) { + policy := defaultRuntimePolicy().Task + if policy.VideoTimeoutMinutes != 60 { + t.Fatalf("default video timeout = %d minutes, want 60", policy.VideoTimeoutMinutes) + } + if got := taskExecutionTimeoutWithPolicy("canvas_video", policy); got != time.Hour { + t.Fatalf("video execution timeout = %s, want 1h", got) + } +} + +func TestProviderPollingDeadlineDefaultsToOneHour(t *testing.T) { + startedAt := time.Now() + deadline := providerPollingDeadline(context.Background()) + remaining := deadline.Sub(startedAt) + if remaining < time.Hour || remaining > time.Hour+time.Second { + t.Fatalf("provider polling deadline = %s, want approximately 1h", remaining) + } +} + func TestOnlyResumableNewAPIChannel2VideoDeadlinesStayRunning(t *testing.T) { svc := &Service{} input, err := json.Marshal(canvasGenerationInput{Mode: "video", Config: providerConfig{BaseURL: "https://example.com", InterfaceType: string(model.ChannelInterfaceNewAPIChannel2)}}) diff --git a/backend/internal/app/task_api_call_log_test.go b/backend/internal/app/task_api_call_log_test.go index c5778953e..2489e779e 100644 --- a/backend/internal/app/task_api_call_log_test.go +++ b/backend/internal/app/task_api_call_log_test.go @@ -4,6 +4,7 @@ import ( "errors" "strings" "testing" + "time" "infinite-canvas/backend/internal/model" "infinite-canvas/backend/internal/repository" @@ -12,6 +13,43 @@ import ( "gorm.io/gorm" ) +func TestVideoPollLogSchedulesNextCheckThirtySecondsLater(t *testing.T) { + db, err := gorm.Open(sqlite.Open("file:provider-poll-schedule?mode=memory&cache=shared"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.Task{}, &model.ApiCallLog{}, &model.ModelPricing{}); err != nil { + t.Fatal(err) + } + task := model.Task{ID: "task-1", UserID: "user-1", Type: "canvas_video", Status: model.TaskStatusRunning} + if err := db.Create(&task).Error; err != nil { + t.Fatal(err) + } + root := model.ApiCallLog{ID: "api-log-1", UserID: task.UserID, TaskID: task.ID, Capability: "video", RequestKind: "create", ProviderRequestID: "provider-task-1", CreatedAt: time.Now()} + if err := db.Create(&root).Error; err != nil { + t.Fatal(err) + } + service := &Service{repo: repository.New(db)} + startedAt := time.Now() + if err := service.LogAPICall(model.ApiCallLog{ + ID: "poll-log-1", UserID: task.UserID, TaskID: task.ID, Capability: "video", RequestKind: "poll", + ProviderRequestID: "provider-task-1", Status: model.ApiCallStatusSucceeded, StatusCode: 200, CreatedAt: startedAt, + }); err != nil { + t.Fatal(err) + } + var stored model.Task + if err := db.First(&stored, "id = ?", task.ID).Error; err != nil { + t.Fatal(err) + } + if stored.NextPollAt == nil { + t.Fatal("next poll time is nil") + } + delay := stored.NextPollAt.Sub(startedAt) + if delay < 30*time.Second || delay > 31*time.Second { + t.Fatalf("next poll delay = %s, want approximately 30s", delay) + } +} + func TestEnsureFailedProviderAttemptLoggedFillsPreflightGapOnce(t *testing.T) { db, err := gorm.Open(sqlite.Open("file:provider-preflight-log?mode=memory&cache=shared"), &gorm.Config{}) if err != nil { diff --git a/backend/internal/app/workflow_provider.go b/backend/internal/app/workflow_provider.go index 919ca5f83..1ecf94bf0 100644 --- a/backend/internal/app/workflow_provider.go +++ b/backend/internal/app/workflow_provider.go @@ -375,7 +375,7 @@ func (s *Service) runRunningHubWorkflow(ctx context.Context, input canvasGenerat root := runningHubRootURL(input.Config.BaseURL) apiKey := runningHubAPIKey(input.Config) if resumed := resumedProviderRequestID(ctx); resumed != "" { - return s.pollRunningHubWorkflow(ctx, input.Config, root, resumed) + return s.pollRunningHubWorkflow(ctx, input.Config, root, resumed, input.Mode) } workflowID := strings.TrimSpace(input.Config.WorkflowID) webappID := strings.TrimSpace(input.Config.WebappID) @@ -459,7 +459,7 @@ func (s *Service) runRunningHubWorkflow(ctx context.Context, input canvasGenerat metadata, _ := ctx.Value(providerAnalyticsKey{}).(providerAnalyticsContext) _ = s.log(metadata.UserID, metadata.TaskID, "error", "RunningHub 请求状态保存失败", taskID+"："+err.Error()) } - return s.pollRunningHubWorkflow(ctx, input.Config, root, taskID) + return s.pollRunningHubWorkflow(ctx, input.Config, root, taskID, input.Mode) } func runningHubWorkflowFailureMessage(response map[string]any) string { @@ -1641,7 +1641,59 @@ func (s *Service) runningHubJSON(ctx context.Context, config providerConfig, end return nil } -func (s *Service) pollRunningHubWorkflow(ctx context.Context, config providerConfig, root string, taskID string) (map[string]interface{}, error) { +func (s *Service) pollRunningHubWorkflow(ctx context.Context, config providerConfig, root string, taskID string, mode string) (map[string]interface{}, error) { + if mode == "video" { + return s.pollRunningHubVideoWorkflowWithPolicy(ctx, config, root, taskID, defaultVideoPollPolicy()) + } + return s.pollRunningHubWorkflowLegacy(ctx, config, root, taskID) +} + +func (s *Service) pollRunningHubVideoWorkflowWithPolicy(ctx context.Context, config providerConfig, root string, taskID string, policy videoPollPolicy) (map[string]interface{}, error) { + return runVideoPollLoop(ctx, taskID, policy, func(ctx context.Context) (videoPollOutcome, error) { + var response map[string]any + err := s.runningHubJSON(withProviderRequestKind(ctx, "poll"), config, root+"/task/openapi/outputs", map[string]any{"apiKey": runningHubAPIKey(config), "taskId": taskID}, &response) + if err != nil { + return videoPollOutcome{}, fmt.Errorf("RunningHub 查询任务失败：%w", err) + } + code, validCode := runningHubPayloadCode(response) + if !validCode { + validCode = len(runningHubOutputURLs(response["data"])) > 0 + code = 0 + } + if !validCode { + return videoPollOutcome{}, errors.New("RunningHub 查询响应缺少可识别状态") + } + if code == 0 { + urls := runningHubOutputURLs(response["data"]) + if len(urls) == 0 { + return videoPollOutcome{}, errors.New("RunningHub 任务成功但没有返回产物") + } + for index, rawURL := range urls { + urls[index] = resolveRunningHubOutputURL(root, rawURL) + } + result, err := s.downloadWorkflowVideoOutputs(ctx, urls, taskID, policy) + if err != nil { + return videoPollOutcome{}, err + } + _ = s.updateWorkflowProviderState(ctx, taskID, "succeeded", nil) + return videoPollOutcome{Done: true, Result: result}, nil + } + if code == 805 || code == 806 { + return videoPollOutcome{}, fmt.Errorf("RunningHub 任务失败：%s", runningHubFailureMessage(response)) + } + stage := "running" + if code == 813 { + stage = "queued" + } else if code != 804 { + stage = "pending" + } + next := time.Now().Add(policy.Interval) + _ = s.updateWorkflowProviderState(ctx, taskID, stage, &next) + return videoPollOutcome{}, nil + }) +} + +func (s *Service) pollRunningHubWorkflowLegacy(ctx context.Context, config providerConfig, root string, taskID string) (map[string]interface{}, error) { for deadline := providerPollingDeadline(ctx); time.Now().Before(deadline); { var response map[string]any err := s.runningHubJSON(withProviderRequestKind(ctx, "poll"), config, root+"/task/openapi/outputs", map[string]any{"apiKey": runningHubAPIKey(config), "taskId": taskID}, &response) @@ -1692,6 +1744,14 @@ func runningHubAPIKey(config providerConfig) string { } func (s *Service) downloadWorkflowOutputs(ctx context.Context, urls []string) (map[string]interface{}, error) { + return s.downloadWorkflowOutputsWithPolicy(ctx, urls, "", nil) +} + +func (s *Service) downloadWorkflowVideoOutputs(ctx context.Context, urls []string, taskID string, policy videoPollPolicy) (map[string]interface{}, error) { + return s.downloadWorkflowOutputsWithPolicy(ctx, urls, taskID, &policy) +} + +func (s *Service) downloadWorkflowOutputsWithPolicy(ctx context.Context, urls []string, taskID string, policy *videoPollPolicy) (map[string]interface{}, error) { images := make([]map[string]interface{}, 0) var video, audio map[string]interface{} for _, rawURL := range urls { @@ -1717,7 +1777,16 @@ func (s *Service) downloadWorkflowOutputs(ctx context.Context, urls []string) (m if !isPublicMediaURL(rawURL) { continue } - data, mimeType, err := getExternalBinary(withProviderRequestKind(ctx, "download"), rawURL) + var data []byte + var mimeType string + var err error + if policy == nil { + data, mimeType, err = getExternalBinary(withProviderRequestKind(ctx, "download"), rawURL) + } else { + data, mimeType, err = runVideoDownload(ctx, taskID, *policy, func(ctx context.Context) ([]byte, string, error) { + return getExternalBinary(withProviderRequestKind(ctx, "download"), rawURL) + }) + } if err != nil { return nil, fmt.Errorf("下载 RunningHub 产物失败：%w", err) } diff --git a/backend/internal/app/workflow_video_polling_test.go b/backend/internal/app/workflow_video_polling_test.go new file mode 100644 index 000000000..28eddf8fd --- /dev/null +++ b/backend/internal/app/workflow_video_polling_test.go @@ -0,0 +1,42 @@ +package app + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" +) + +func TestRunningHubVideoPollRecoversFromTransientGatewayFailure(t *testing.T) { + allowLoopbackProviderTest(t) + pollCalls := 0 + var server *httptest.Server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/task/openapi/outputs": + pollCalls++ + w.Header().Set("Content-Type", "application/json") + if pollCalls == 1 { + w.WriteHeader(http.StatusBadGateway) + return + } + _, _ = w.Write([]byte(`{"code":0,"data":[{"fileUrl":"` + server.URL + `/result.mp4"}]}`)) + case "/result.mp4": + w.Header().Set("Content-Type", "video/mp4") + _, _ = w.Write([]byte("video")) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + config := providerConfig{BaseURL: server.URL, APIKey: "test-key"} + ctx := context.Background() + result, err := (&Service{}).pollRunningHubVideoWorkflowWithPolicy(ctx, config, server.URL, "provider-task-1", fastVideoPollPolicy()) + if err != nil { + t.Fatal(err) + } + if result["mode"] != "video" || pollCalls != 2 { + t.Fatalf("result = %#v, poll calls = %d", result, pollCalls) + } +} diff --git a/backend/internal/platform/runtime_policy.go b/backend/internal/platform/runtime_policy.go index 23f947d73..7e5514dc6 100644 --- a/backend/internal/platform/runtime_policy.go +++ b/backend/internal/platform/runtime_policy.go @@ -120,7 +120,7 @@ func DefaultRuntimePolicy() RuntimePolicySetting { ImageTimeoutMinutes: 8, TextTimeoutMinutes: 8, AudioTimeoutMinutes: 8, - VideoTimeoutMinutes: 30, + VideoTimeoutMinutes: 60, StoryboardTimeoutMinutes: 20, DefaultTimeoutMinutes: 10, }, diff --git a/docs/content/docs/backend/newapi-channel-2.mdx b/docs/content/docs/backend/newapi-channel-2.mdx index ff3eb2f2c..f136e1f81 100644 --- a/docs/content/docs/backend/newapi-channel-2.mdx +++ b/docs/content/docs/backend/newapi-channel-2.mdx @@ -26,26 +26,18 @@ description: Channel 2 视频任务创建、轮询与上游任务同步延迟处 宿主只在同时满足以下条件时把该响应视为“任务状态尚未同步”： -1. 渠道接口类型精确为 `newapi-channel-2`； -2. 已经保存非空 Provider Task ID； -3. HTTP 状态为 `400`； -4. JSON 的 `code` 或 `message` 精确等于 `task_not_exist`。 +1. 已经保存非空 Provider Task ID； +2. HTTP 状态为 `400` 或 `404`； +3. JSON 的 `code` 或 `message` 精确等于 `task_not_exist`、`task_not_found` 或对应的空格拼写。 -不符合条件的 HTTP 错误、非 JSON 正文或其他错误码仍立即按失败语义处理。 +该判断属于宿主统一的视频轮询语义，不再按渠道名分支。不符合条件的 400/401/403/422 等参数、鉴权或权限错误仍立即按失败语义处理。 ## 轮询与延期 -新任务首次查询前等待 `2` 秒；同步错误在同一次执行中按 `2` 秒、`3` 秒、`5` 秒、`5` 秒退避，最多短期查询 5 次。短期窗口耗尽后，Worker 不发送新的创建请求，也不切换渠道或退款，而是： +新任务和恢复任务都在首次查询前等待 30 秒，后续统一每 30 秒查询。连续三次 `task_not_exist` 或 404 才结束；中间任何一次正常的 pending/running 响应都会清零连续计数。HTTP 408/409/425/429/5xx、临时网络错误、断流、渠道槽位和短时熔断继续查询原 Provider Task ID，不发送第二次创建请求。 -- 保持本地任务状态为 `running`； -- 保留原 `provider_request_id` 和计费状态； -- 释放当前租约并把 `next_poll_at` 延后 15 秒； -- 下一轮从已保存的 Provider Task ID 恢复，只发送查询请求。 - -`task_not_exist` 的跨轮次同步等待从本地任务首次开始执行起最多持续 5 分钟。超过该上限仍未得到有效任务状态时，Worker 将任务置为失败并释放活跃任务与计费预留，避免无效 Provider Task ID 永久占用用户任务名额。该限制只针对持续的 `task_not_exist`；查询接口已经返回正常 pending/running 状态后的长视频生成仍按视频任务超时和延期策略恢复。 - -因此，上游生成耗时 500 至 600 秒时，只要查询接口返回有效的处理中状态，本地会持续以低频查询等待完成，不会把正常排队误报为“模型服务拒绝了请求”。视频真正返回失败状态时，仍按供应商失败处理。 +默认视频执行窗口为 60 分钟；管理员已经保存的运行时策略仍按保存值执行。视频真正返回 failed/cancelled/expired 等终态时，仍按供应商失败处理。 ## 兼容边界 -该容错只适用于 `newapi-channel-2` 视频任务。其他 NewAPI 接口、其他 Provider、缺少 Provider Task ID 的任务，以及真实参数错误不进入延期路径。 +该容错适用于所有异步视频声明式插件和宿主内置协议。缺少 Provider Task ID 的创建失败、真实参数错误、鉴权错误和非法插件映射不进入重试路径。 diff --git a/docs/content/docs/progress/pending-test.mdx b/docs/content/docs/progress/pending-test.mdx index 2a97c18b0..656dfe3fa 100644 --- a/docs/content/docs/progress/pending-test.mdx +++ b/docs/content/docs/progress/pending-test.mdx @@ -78,6 +78,18 @@ description: 当前项目已实现功能的手工验证清单与未确认边界 - [ ] 登录真实画布验收入口、设置、详情、输入及窗口偏好；检查画布缩放和快捷键不干扰浮窗。 - [ ] 真实 Agent 运行、工具写入、媒体生成与 SSE 重连待联调；当前专项测试使用模拟模型和本地数据库，不替代真实上游验收。 +## 异步视频统一轮询容错 + +- 声明式视频插件、遗留内置视频协议和 RunningHub 视频工作流统一在创建成功后等待 30 秒首次查询，后续每 30 秒查询；图片、音频和文本协议不受影响。 +- 查询阶段的 HTTP 408/409/425/429/5xx、临时网络错误、断流、渠道槽位或熔断，以及最多连续两次非法 JSON 不再直接结束任务；HTTP 404 和结构化 `task_not_exist` 连续第三次才结束。 +- 结果下载复用同一暂态错误分类，最多尝试 3 次且不重新创建或轮询上游任务。默认视频任务执行窗口由 30 分钟改为 60 分钟；已保存的管理员策略不会被覆盖。 +- [x] `cd backend && go test ./internal/app ./internal/platform -count=1`：通过。 +- [x] `cd backend && go test ./... -count=1`、`go vet ./...`：通过。 +- [ ] 在测试渠道制造一次查询 502，确认任务保持运行、供应商任务 ID 不变，后续查询恢复后正常完成且不重复扣费。 +- [ ] 检查任务日志只出现一次「查询暂时异常」与一次「查询已恢复」，请求明细仍保留每次上游调用。 +- [ ] 让视频下载首次返回 502、第二次成功，确认宿主不重新查询或创建任务；连续三次下载失败时应返回包含供应商任务 ID 的错误。 +- [ ] 将管理员视频超时仍设为旧值的升级环境启动一次，确认保存值保持不变；新安装环境默认显示 60 分钟。 + ## 画布图片资源缓存链路优化 消除首屏满屏转圈与"一直在下载"：后端图片资源改 30 天磁盘强缓存（`Cache-Control: max-age=2592000`），前端内存防抖 IndexedDB 元数据写入、O(1) 淘汰预检、并发下载 4→16，视口内远程图片直链首帧上屏（走原生磁盘缓存秒开），Blob 缓存后台平滑升级。已编译通过，需浏览器最终确认： diff --git a/docs/superpowers/specs/2026-09-12-unified-video-polling-design.md b/docs/superpowers/specs/2026-09-12-unified-video-polling-design.md new file mode 100644 index 000000000..81b986c70 --- /dev/null +++ b/docs/superpowers/specs/2026-09-12-unified-video-polling-design.md @@ -0,0 +1,109 @@ +# 统一异步视频轮询容错设计 + +## 背景 + +当前异步视频渠道分别在声明式协议和多个内置协议中维护轮询循环。轮询间隔分散为 2.5 秒、3 秒、5 秒、10 秒或 20 秒；除少数协议外，一次 HTTP 502、429、网络超时或连接中断就会结束执行，使本地任务失败并退款，而上游任务可能仍在继续并最终成功。 + +本改动面向开源宿主整体修复，不识别任何供应商、域名或私有模型。 + +## 目标 + +- 所有异步视频协议统一每 30 秒轮询一次，声明式插件和内置协议行为一致。 +- 创建成功后等待 30 秒再首次查询，避免刚创建即查询。 +- 单次可恢复查询异常不得终止整个生成任务。 +- 只在上游明确返回终态失败、遇到确定不可恢复错误或达到任务总超时时结束任务。 +- 成品已生成后，临时下载异常不得立即丢失任务结果。 +- 保留现有任务 ID、计费、审计、渠道并发、熔断和人工恢复边界。 + +## 非目标 + +- 不改变各供应商的请求字段、状态映射或结果结构。 +- 不为单独渠道增加判断或插件私有配置。 +- 不将轮询改造成释放 worker 的持久化后台状态机；该方向可作为后续独立重构。 +- 不自动重试创建请求，避免在缺少供应商幂等保证时重复创建计费任务。 + +## 统一轮询执行器 + +在 `backend/internal/app` 增加宿主级视频轮询执行器。所有异步视频实现通过回调完成一次查询和供应商特有的状态解析，执行器负责等待、错误分类、重试、超时和通用日志。 + +执行流程： + +1. 创建请求成功并取得供应商任务 ID。 +2. 等待 30 秒。 +3. 执行一次查询。 +4. 查询返回处理中时，再等待 30 秒。 +5. 查询遇到可恢复异常时记录重试信息，等待统一间隔或更长的 `Retry-After` 后继续。 +6. 查询成功后清零连续异常计数。 +7. 收到成功或明确失败终态时退出。 +8. 上下文截止时按视频任务超时处理。 + +恢复一个已有供应商任务 ID 时也遵守同一轮询节奏，避免 worker 重启后形成突发查询。 + +## 错误分类 + +以下只适用于已经取得供应商任务 ID 后的查询或结果下载阶段： + +### 可恢复 + +- HTTP 408、409、425、429、500、502、503、504 以及其他 5xx。 +- DNS、连接重置、临时网络错误、请求超时和响应读取中断。 +- 渠道熔断暂时开启或并发槽暂时不可用。 +- 查询返回 200 但响应暂时不是合法 JSON：允许有限次数连续重试，避免网关临时返回 HTML；连续 3 次后视为协议错误。 + +HTTP 429 的 `Retry-After` 大于 30 秒时优先遵守，但等待时间不得越过任务截止时间。 + +### 有条件恢复 + +- HTTP 404：连续 3 次查询都不存在后才结束，兼容供应商创建后异步同步任务记录的窗口。 +- 已有协议明确把某个业务错误定义为“尚不可查”时，统一纳入相同计数，不再维护供应商专属睡眠循环。 + +### 不可恢复 + +- HTTP 400、401、403、422 等明确的参数、鉴权或权限错误。 +- 插件清单无法构造请求、响应映射本身非法或本地校验失败。 +- 上游明确返回 `failed`、`cancelled`、`canceled`、`expired` 等终态。 +- 用户主动取消或宿主关闭导致的上下文取消。 + +## 下载阶段 + +结果状态成功后，媒体下载复用同一可恢复网络错误分类。下载最多连续尝试 3 次，每次至少等待 30 秒并遵守 `Retry-After`；仍失败时返回带供应商任务 ID 的明确错误，人工恢复仍可只查询原任务并再次下载，不创建新任务。 + +## 超时与运行时策略 + +- 全局视频任务默认超时从 30 分钟调整为 60 分钟。 +- 已保存的管理员运行时策略保持原值，升级不静默覆盖管理员配置；管理员可手动调整为 60 分钟。 +- 所有等待和 HTTP 请求都服从任务上下文截止时间，不额外延长租约之外的生命周期。 + +## 日志与计费 + +- 每次上游 HTTP 失败仍保留 API 调用审计记录。 +- 可恢复的轮询异常不得触发任务终态、退款或重新创建供应商任务。 +- 首次进入重试和恢复正常时写任务日志；连续相同错误不按每次轮询刷屏。 +- 最终成功沿用现有结算流程；明确失败或总超时沿用现有失败与退款流程。 + +## 兼容与迁移 + +- 不修改数据库结构。 +- 声明式插件清单无需升级，已安装插件自动获得统一策略。 +- 内置视频协议只迁移其查询循环，不改变创建和响应解析代码。 +- 前端仍查询宿主任务状态，无需改变前端轮询间隔。 + +## 测试策略 + +先写失败测试，再实现： + +- 统一间隔常量为 30 秒，首次查询前也等待该间隔。 +- 声明式协议经历 `502 -> processing -> succeeded` 后成功，且只创建一次上游任务。 +- 一个内置协议经历网络超时或 503 后继续并成功。 +- 401、400 和明确失败状态立即结束。 +- 404 前两次重试、第三次结束；正常响应会重置连续错误计数。 +- `Retry-After` 生效且不越过上下文截止时间。 +- 下载在临时 502 后重试成功，不重复创建任务。 +- 上下文取消与总超时能够及时退出。 +- 后端完整测试 `go test ./...` 通过。 + +## 风险与控制 + +- 统一 30 秒会降低原来高频协议的结果发现速度，但显著减少供应商压力和限流概率；这是本次明确接受的产品取舍。 +- 持续故障会让任务保持运行更久，并占用 worker。通过现有并发上限、渠道熔断和 60 分钟总超时控制；后台状态机另行设计。 +- 过宽重试可能掩盖配置错误，因此鉴权、参数、映射和明确终态不进入重试。