diff --git a/backend/internal/app/admin.go b/backend/internal/app/admin.go index 1903e91bd..d374fa009 100644 --- a/backend/internal/app/admin.go +++ b/backend/internal/app/admin.go @@ -664,10 +664,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 d21fd9bde..816e48f0b 100644 --- a/backend/internal/app/protocol_runtime_test.go +++ b/backend/internal/app/protocol_runtime_test.go @@ -13,6 +13,7 @@ import ( "regexp" "strings" "testing" + "time" "infinite-canvas/backend/internal/protocol" ) @@ -348,6 +349,18 @@ 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) { adapter, err := protocol.LoadManifest([]byte(`{ "apiVersion":"yingce.plugin/v1", diff --git a/backend/internal/app/provider.go b/backend/internal/app/provider.go index f3cf5e7e8..ddbdfbc86 100644 --- a/backend/internal/app/provider.go +++ b/backend/internal/app/provider.go @@ -100,7 +100,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 { @@ -145,6 +145,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 350c81d37..7003694ef 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 != "" { @@ -269,7 +269,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 ab48f51a5..8acfc0d68 100644 --- a/backend/internal/app/provider_protocol.go +++ b/backend/internal/app/provider_protocol.go @@ -40,7 +40,19 @@ func runDeclarativeProtocolTask(ctx context.Context, input canvasGenerationInput } func runProtocolAdapterTask(ctx context.Context, input canvasGenerationInput, adapter protocol.Adapter) (map[string]interface{}, error) { - return runProtocolAdapterTaskWithPolicy(ctx, input, adapter, defaultVideoPollPolicy()) + return runProtocolAdapterTaskWithPolicy(ctx, input, adapter, declarativeProtocolPollPolicy(input.Mode)) +} + +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) { diff --git a/backend/internal/app/provider_video_polling.go b/backend/internal/app/provider_video_polling.go index e542bd859..ce404f619 100644 --- a/backend/internal/app/provider_video_polling.go +++ b/backend/internal/app/provider_video_polling.go @@ -4,6 +4,8 @@ import ( "context" "encoding/json" "errors" + "fmt" + "io" "net" "net/http" "strings" @@ -13,54 +15,111 @@ import ( const defaultVideoPollInterval = 30 * time.Second type videoPollPolicy struct { - Interval time.Duration - MaxNotFoundMisses int - MaxDownloadTries int - Sleep func(context.Context, time.Duration) error + 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{ - Interval: defaultVideoPollInterval, - MaxNotFoundMisses: 3, - MaxDownloadTries: 3, - Sleep: sleepContext, + 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) - nextDelay := policy.Interval + 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 err := policy.Sleep(ctx, nextDelay); err != nil { - return nil, err + 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 } - if notFound { + 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 } @@ -78,12 +137,18 @@ func normalizeVideoPollPolicy(policy videoPollPolicy) videoPollPolicy { 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 } @@ -98,23 +163,37 @@ func runVideoDownload(ctx context.Context, taskID string, policy videoPollPolicy lastErr = err retry, _ := retryableVideoPollError(ctx, err) if !retry || attempt == policy.MaxDownloadTries { - return nil, "", err + 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, "", lastErr + 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 } @@ -140,6 +219,33 @@ func retryableVideoPollError(ctx context.Context, err error) (retry bool, notFou 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 diff --git a/backend/internal/app/provider_video_polling_test.go b/backend/internal/app/provider_video_polling_test.go index 7027237fb..00c8c8604 100644 --- a/backend/internal/app/provider_video_polling_test.go +++ b/backend/internal/app/provider_video_polling_test.go @@ -2,7 +2,9 @@ package app import ( "context" + "encoding/json" "errors" + "io" "net/http" "reflect" "testing" @@ -33,6 +35,7 @@ func TestVideoPollRecoversFromTransientHTTPError(t *testing.T) { 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) @@ -105,6 +108,68 @@ func TestVideoPollSuccessfulPendingResponseResetsNotFoundCounter(t *testing.T) { } } +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() @@ -121,6 +186,26 @@ func TestVideoPollContextCancellationInterruptsInitialWait(t *testing.T) { } } +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 @@ -156,8 +241,21 @@ func TestVideoDownloadStopsAfterThreeTransientFailures(t *testing.T) { } } +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 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 6b33ea21d..7b6deca22 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 4bade3bf9..ff7a12330 100644 --- a/backend/internal/app/workflow_provider.go +++ b/backend/internal/app/workflow_provider.go @@ -451,7 +451,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) @@ -535,7 +535,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 { @@ -1717,7 +1717,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) @@ -1768,6 +1820,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 { @@ -1793,7 +1853,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..611b1043d --- /dev/null +++ b/backend/internal/app/workflow_video_polling_test.go @@ -0,0 +1,41 @@ +package app + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" +) + +func TestRunningHubVideoPollRecoversFromTransientGatewayFailure(t *testing.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", AllowLocalChannel: true} + ctx := withProviderOutboundPolicy(context.Background(), config) + 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 ca7386e37..ba91a3281 100644 --- a/backend/internal/platform/runtime_policy.go +++ b/backend/internal/platform/runtime_policy.go @@ -127,7 +127,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 50d4fe29a..844525112 100644 --- a/docs/content/docs/progress/pending-test.mdx +++ b/docs/content/docs/progress/pending-test.mdx @@ -5,6 +5,18 @@ description: 当前项目已实现功能的手工验证清单与未确认边界 # 待测试 +## 异步视频统一轮询容错 + +- 声明式视频插件、遗留内置视频协议和 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 缓存后台平滑升级。已编译通过，需浏览器最终确认：