tt/ttstd-infinite-canvas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(video): 结果下载 - 暂态失败不重复轮询任务

Browse Source
This commit is contained in:
hzw
2026-09-12 21:10:48 +08:00
parent 2df5bcc5cd
commit d2efd81b8e
6 changed files with 204 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
backend/internal/app/protocol_runtime_test.go
View File

@@ -395,6 +395,52 @@ func TestDeclarativeProtocolPollRecoversFromTransientGatewayFailure(t *testing.T
}
}
func TestDeclarativeProtocolRetriesResultDownloadWithoutRepolling(t *testing.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", 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" || pollCalls != 1 || downloadCalls != 2 {
t.Fatalf("result = %#v, poll calls = %d, download calls = %d", result, pollCalls, downloadCalls)
}
}
func TestDeclarativeProtocolRuntimeGeneratesPerCreateIdempotencyKey(t *testing.T) {
manifest := []byte(`{
"apiVersion":"yingce.plugin/v1",

63
backend/internal/app/provider_protocol.go
View File

@@ -77,7 +77,7 @@ func runProtocolAdapterTaskWithPolicy(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")
@@ -102,7 +102,7 @@ func runProtocolAdapterTaskWithPolicy(ctx context.Context, input canvasGeneratio
}
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, policy)
return videoPollOutcome{Done: err == nil, Result: result}, err
case protocol.StatusFailed, protocol.StatusCancelled:
return videoPollOutcome{}, protocolResultError(state.Message, taskID)
@@ -152,7 +152,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)
@@ -731,7 +731,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("声明式协议已完成但没有返回结果")
}
@@ -758,7 +758,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
}
@@ -778,20 +787,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)
}
@@ -814,7 +832,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 {
@@ -836,21 +854,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
@@ -862,6 +872,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

39
backend/internal/app/provider_test.go
View File

@@ -1625,6 +1625,45 @@ func TestRunVideoTaskRetriesTransientPollFailureWithoutRecreating(t *testing.T)
}
}
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

20
backend/internal/app/provider_video.go
View File

@@ -145,14 +145,18 @@ func runVideoTaskWithPolicy(ctx context.Context, input canvasGenerationInput, po
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 videoPollOutcome{}, fmt.Errorf("视频结果下载失败（任务 %s）：%w", id, err)
}
mimeType = normalizedMediaMimeType(mimeType, data)
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 videoPollOutcome{}, err
}
@@ -258,13 +262,17 @@ func runSeedanceVideosTask(ctx context.Context, input canvasGenerationInput, pol
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 videoPollOutcome{}, fmt.Errorf("视频结果下载失败：%w", err)
}
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 videoPollOutcome{}, fmt.Errorf("Seedance 任务成功但未返回视频 URL，备用内容下载失败：%w", err)
}
@@ -343,7 +351,9 @@ func runSeedanceAgentPlanVideoTask(ctx context.Context, input canvasGenerationIn
if videoURL == "" {
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 videoPollOutcome{}, fmt.Errorf("视频结果下载失败：%w", err)
}

26
backend/internal/app/provider_video_polling.go
View File

@@ -15,6 +15,7 @@ const defaultVideoPollInterval = 30 * time.Second
type videoPollPolicy struct {
Interval time.Duration
MaxNotFoundMisses int
MaxDownloadTries int
Sleep func(context.Context, time.Duration) error
}
@@ -27,6 +28,7 @@ func defaultVideoPollPolicy() videoPollPolicy {
return videoPollPolicy{
Interval: defaultVideoPollInterval,
MaxNotFoundMisses: 3,
MaxDownloadTries: 3,
Sleep: sleepContext,
}
}
@@ -76,12 +78,36 @@ func normalizeVideoPollPolicy(policy videoPollPolicy) videoPollPolicy {
if policy.MaxNotFoundMisses <= 0 {
policy.MaxNotFoundMisses = 3
}
if policy.MaxDownloadTries <= 0 {
policy.MaxDownloadTries = 3
}
if policy.Sleep == nil {
policy.Sleep = sleepContext
}
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, "", err
}
delay := max(policy.Interval, providerRetryAfter(err))
if err := policy.Sleep(ctx, delay); err != nil {
return nil, "", err
}
}
return nil, "", 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

35
backend/internal/app/provider_video_polling_test.go
View File

@@ -121,6 +121,41 @@ func TestVideoPollContextCancellationInterruptsInitialWait(t *testing.T) {
}
}
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 fastVideoPollPolicy() videoPollPolicy {
policy := defaultVideoPollPolicy()
policy.Interval = time.Millisecond