feat(video): 异步任务 - 增加统一轮询容错执行器

backend/internal/app/provider_video_polling.go

@@ -0,0 +1,118 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"net"
+	"net/http"
+	"time"
+)
+
+const defaultVideoPollInterval = 30 * time.Second
+
+type videoPollPolicy struct {
+	Interval          time.Duration
+	MaxNotFoundMisses int
+	Sleep             func(context.Context, time.Duration) error
+}
+
+type videoPollOutcome struct {
+	Done   bool
+	Result map[string]interface{}
+}
+
+func defaultVideoPollPolicy() videoPollPolicy {
+	return videoPollPolicy{
+		Interval:          defaultVideoPollInterval,
+		MaxNotFoundMisses: 3,
+		Sleep:             sleepContext,
+	}
+}
+
+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
+	notFoundMisses := 0
+	for time.Now().Before(deadline) {
+		if err := policy.Sleep(ctx, nextDelay); err != nil {
+			return nil, err
+		}
+		outcome, err := query(ctx)
+		if err != nil {
+			retry, notFound := retryableVideoPollError(ctx, err)
+			if !retry {
+				return nil, err
+			}
+			if notFound {
+				notFoundMisses++
+				if notFoundMisses >= policy.MaxNotFoundMisses {
+					return nil, err
+				}
+			} else {
+				notFoundMisses = 0
+			}
+			nextDelay = max(policy.Interval, providerRetryAfter(err))
+			continue
+		}
+		notFoundMisses = 0
+		nextDelay = policy.Interval
+		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.Sleep == nil {
+		policy.Sleep = sleepContext
+	}
+	return policy
+}
+
+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
+	}
+	if code, _ := ChannelSlotFailureDetails(err); code != "" {
+		return true, false
+	}
+	if errors.Is(err, context.DeadlineExceeded) {
+		return true, false
+	}
+	var httpErr providerHTTPError
+	if errors.As(err, &httpErr) {
+		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 providerRetryAfter(err error) time.Duration {
+	var httpErr providerHTTPError
+	if errors.As(err, &httpErr) && httpErr.RetryAfter > 0 {
+		return httpErr.RetryAfter
+	}
+	return 0
+}

backend/internal/app/provider_video_polling_test.go

@@ -0,0 +1,129 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"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.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 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 fastVideoPollPolicy() videoPollPolicy {
+	policy := defaultVideoPollPolicy()
+	policy.Interval = time.Millisecond
+	policy.Sleep = func(context.Context, time.Duration) error { return nil }
+	return policy
+}