From a2cd908567a138e554321062048d8f73b9c658a5 Mon Sep 17 00:00:00 2001 From: hzw Date: Sat, 12 Sep 2026 21:02:01 +0800 Subject: [PATCH] =?UTF-8?q?fix(video):=20=E5=A3=B0=E6=98=8E=E5=BC=8F?= =?UTF-8?q?=E5=8D=8F=E8=AE=AE=20-=20=E7=BB=9F=E4=B8=80=E8=BD=AE=E8=AF=A2?= =?UTF-8?q?=E5=B9=B6=E5=AE=B9=E5=BF=8D=E6=9A=82=E6=80=81=E9=94=99=E8=AF=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/app/protocol_runtime_test.go | 112 ++++++++++++------ backend/internal/app/provider_protocol.go | 90 ++------------ .../internal/app/provider_video_polling.go | 23 ++++ 3 files changed, 110 insertions(+), 115 deletions(-) diff --git a/backend/internal/app/protocol_runtime_test.go b/backend/internal/app/protocol_runtime_test.go index 26eb96e57..6e823a283 100644 --- a/backend/internal/app/protocol_runtime_test.go +++ b/backend/internal/app/protocol_runtime_test.go @@ -13,7 +13,6 @@ import ( "regexp" "strings" "testing" - "time" "infinite-canvas/backend/internal/protocol" ) @@ -336,7 +335,11 @@ func TestDeclarativeProtocolRuntimeExecutesCreatePollAndDownload(t *testing.T) { config := providerConfig{BaseURL: server.URL + "/v1", APIKey: "key", Model: "test-model", APIFormat: "openai", InterfaceType: "test-declarative-video-runtime", AllowLocalChannel: true} ctx := withProviderOutboundPolicy(context.Background(), config) 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) } @@ -345,6 +348,53 @@ func TestDeclarativeProtocolRuntimeExecutesCreatePollAndDownload(t *testing.T) { } } +func TestDeclarativeProtocolPollRecoversFromTransientGatewayFailure(t *testing.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", AllowLocalChannel: true} + ctx := withProviderOutboundPolicy(context.Background(), config) + 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 TestDeclarativeProtocolRuntimeGeneratesPerCreateIdempotencyKey(t *testing.T) { manifest := []byte(`{ "apiVersion":"yingce.plugin/v1", @@ -393,8 +443,12 @@ func TestDeclarativeProtocolRuntimeGeneratesPerCreateIdempotencyKey(t *testing.T config := providerConfig{BaseURL: server.URL + "/v1", APIKey: "key", Model: "test-model", APIFormat: "openai", InterfaceType: "test-idempotency-key-runtime", AllowLocalChannel: true} ctx := withProviderOutboundPolicy(context.Background(), config) 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) } @@ -438,12 +492,7 @@ func TestDeclarativeNewAPIChannel2TaskNotExistRetry(t *testing.T) { config := providerConfig{BaseURL: server.URL + "/v1", APIKey: "key", Model: "video-model", InterfaceType: "newapi-channel-2", AllowLocalChannel: true} ctx := withProviderOutboundPolicy(context.Background(), config) - 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) } @@ -475,43 +524,34 @@ func TestDeclarativeNewAPIChannel2TaskNotExistExhaustion(t *testing.T) { config := providerConfig{BaseURL: server.URL + "/v1", APIKey: "key", Model: "video-model", InterfaceType: "newapi-channel-2", AllowLocalChannel: true} ctx := withProviderOutboundPolicy(context.Background(), config) - _, 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_protocol.go b/backend/internal/app/provider_protocol.go index 04f27b00a..bf4914092 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,16 @@ 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, defaultVideoPollPolicy()) } -func runProtocolAdapterTaskWithTiming(ctx context.Context, input canvasGenerationInput, adapter protocol.Adapter, timing protocolPollTiming) (map[string]interface{}, error) { +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 分支。 @@ -101,66 +82,33 @@ func runProtocolAdapterTaskWithTiming(ctx context.Context, input canvasGeneratio 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) + 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) { @@ -184,22 +132,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) { diff --git a/backend/internal/app/provider_video_polling.go b/backend/internal/app/provider_video_polling.go index c9100f5c4..6f72c646f 100644 --- a/backend/internal/app/provider_video_polling.go +++ b/backend/internal/app/provider_video_polling.go @@ -2,9 +2,11 @@ package app import ( "context" + "encoding/json" "errors" "net" "net/http" + "strings" "time" ) @@ -92,6 +94,9 @@ func retryableVideoPollError(ctx context.Context, err error) (retry bool, notFou } var httpErr providerHTTPError if errors.As(err, &httpErr) { + if isProviderTaskNotReadyError(httpErr) { + return true, true + } if httpErr.StatusCode == http.StatusNotFound { return true, true } @@ -109,6 +114,24 @@ func retryableVideoPollError(ctx context.Context, err error) (retry bool, notFou return false, false } +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 {