274 lines
7.3 KiB
Go
274 lines
7.3 KiB
Go
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
|
|
}
|