diff --git a/backend/internal/app/provider_video_polling.go b/backend/internal/app/provider_video_polling.go new file mode 100644 index 000000000..c9100f5c4 --- /dev/null +++ b/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 +} diff --git a/backend/internal/app/provider_video_polling_test.go b/backend/internal/app/provider_video_polling_test.go new file mode 100644 index 000000000..6af234654 --- /dev/null +++ b/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 +}