From 627a5b2d5ed033c1dfbfc367ad4d7bb61fe66be0 Mon Sep 17 00:00:00 2001 From: 404oops Date: Thu, 3 Sep 2026 16:47:16 +0200 Subject: [PATCH] fix: stop video generation timeouts and failures from paging (PUT-1620) A video job that outlived its poll window, an SDK request that timed out, or a Veo operation that finished with an error all reached the HTTP error handler as plain Errors. Each became an unhandled 500 with critical severity and paged on-call for what is the provider's pace or the provider's fault. Video providers now share one poll loop that gives up with a 504 `upstream_timeout`, treats a transient poll failure (timeout, dropped connection, 408/429/5xx) as a missed poll rather than a failed job, and stops polling with a 400 `client_aborted` when the caller disconnects, so nothing is metered for a clip nobody will receive. The driver controller exposes the disconnect as an `abortSignal` on the request context. The window is ten minutes for every provider; Together and BytePlus move up from five. Failed jobs are classified: content-filter refusals become a 400 `bad_request` with `errorCode: moderation_flagged`, rejected parameters a 400 `upstream_bad_request`, and anything else a 502 `upstream_failed`, each carrying the provider's own code. Veo's filtered output keeps `disallowed_value` and gains the same `errorCode`. The sanitizer and content-filter pattern move from the Replicate provider into a shared util so image and video agree. Status-less SDK connection timeouts are translated to a 504 `upstream_timeout` at the driver boundary, and the chat driver records them per attempt so an all-timeout chain is a 504 and a mixed chain is `upstream_failed` instead of an `internal_error` 500. The Together chat client gets the same ten-minute request timeout as the other providers. The OpenAI video provider is left alone beyond an import path: its API is scheduled to shut down on 2026-09-24. Co-Authored-By: Claude Fable 5.1 --- .../DriverController.concurrent.test.ts | 16 +- .../drivers/DriverController.errors.test.ts | 79 ++++- .../controllers/drivers/DriverController.ts | 26 +- src/backend/core/context.ts | 6 + .../ChatCompletionDriver.routing.test.ts | 48 +++ .../ai-chat/ChatCompletionDriver.test.ts | 30 ++ .../drivers/ai-chat/ChatCompletionDriver.ts | 17 +- .../together/TogetherAIProvider.test.ts | 5 +- .../providers/together/TogetherAIProvider.ts | 3 + .../ReplicateImageGenerationProvider.ts | 23 +- .../byteplus/BytePlusVideoProvider.test.ts | 60 +++- .../byteplus/BytePlusVideoProvider.ts | 47 ++- .../gemini/GeminiVideoProvider.test.ts | 76 ++++- .../providers/gemini/GeminiVideoProvider.ts | 49 +-- .../openai/OpenAIVideoProvider.test.ts | 36 +++ .../providers/openai/OpenAIVideoProvider.ts | 5 +- .../ai-video/providers/polling.test.ts | 297 ++++++++++++++++++ .../drivers/ai-video/providers/polling.ts | 163 ++++++++++ .../together/TogetherVideoProvider.test.ts | 145 ++++++++- .../together/TogetherVideoProvider.ts | 57 ++-- .../drivers/util/upstreamErrors.test.ts | 129 ++++++++ src/backend/drivers/util/upstreamErrors.ts | 129 ++++++++ src/docs/src/AI/txt2vid.md | 14 + 23 files changed, 1334 insertions(+), 126 deletions(-) create mode 100644 src/backend/drivers/ai-video/providers/polling.test.ts create mode 100644 src/backend/drivers/ai-video/providers/polling.ts create mode 100644 src/backend/drivers/util/upstreamErrors.test.ts create mode 100644 src/backend/drivers/util/upstreamErrors.ts diff --git a/src/backend/controllers/drivers/DriverController.concurrent.test.ts b/src/backend/controllers/drivers/DriverController.concurrent.test.ts index 0a053ad93..58dcb8864 100644 --- a/src/backend/controllers/drivers/DriverController.concurrent.test.ts +++ b/src/backend/controllers/drivers/DriverController.concurrent.test.ts @@ -263,19 +263,24 @@ describe('DriverController — concurrent acquire/release', () => { }); it('does not attach release listeners when the driver declares no concurrent config', async () => { - // The optimisation that lets the existing test stubs in - // DriverController.test.ts get away without an EventEmitter-shaped - // `res`: skip the once() wiring entirely when there's no spec. + // With no spec there is nothing to release, so the gate must not + // wire `finish`/`close` release listeners. The client-disconnect + // hook is separate and always present: exactly one `close` listener. const driver = makeSyntheticDriver(); (driver as { concurrent?: unknown }).concurrent = undefined; const controller = buildController(driver); const handler = captureCallHandler(controller); - // Bare object with no event-emitter surface — exposes the bug - // case where the gate would try to call `res.once`. + // Minimal `res` that only records which events get listeners. + const listened: string[] = []; const bareRes = { statusCode: 200, body: undefined as unknown, + writableFinished: false, + once(event: string) { + listened.push(event); + return this; + }, status(code: number) { this.statusCode = code; return this; @@ -301,5 +306,6 @@ describe('DriverController — concurrent acquire/release', () => { bareRes as unknown as Response, ); expect(bareRes.body).toMatchObject({ success: true, result: 'pong' }); + expect(listened).toEqual(['close']); }); }); diff --git a/src/backend/controllers/drivers/DriverController.errors.test.ts b/src/backend/controllers/drivers/DriverController.errors.test.ts index 812b01e83..fd9b3db9d 100644 --- a/src/backend/controllers/drivers/DriverController.errors.test.ts +++ b/src/backend/controllers/drivers/DriverController.errors.test.ts @@ -35,9 +35,10 @@ /* eslint-disable @typescript-eslint/no-explicit-any */ import { Readable, Writable } from 'node:stream'; import type { Request, RequestHandler, Response } from 'express'; +import { APIConnectionTimeoutError } from 'openai'; import { beforeEach, describe, expect, it, vi } from 'vitest'; import type { DriverMethodLifecycleEvent } from '../../clients/event/types.js'; -import { runWithContext } from '../../core/context.js'; +import { Context, runWithContext } from '../../core/context.js'; import { configureRateLimit } from '../../core/http/middleware/rateLimit.js'; import { DriverController } from './DriverController.js'; @@ -325,6 +326,32 @@ describe('DriverController upstream error translation', () => { expect(err).toBe('a bare string'); }); + it('maps an SDK connection timeout, which carries no status, to a 504 upstream_timeout', async () => { + const raw = new APIConnectionTimeoutError(); + const { err } = await callWith(throwing(raw)); + expect(err).toMatchObject({ + statusCode: 504, + legacyCode: 'upstream_timeout', + message: 'AI provider timed out', + cause: raw, + }); + }); + + it('maps a fetch timeout that undici wraps in a `fetch failed` TypeError to a 504 upstream_timeout', async () => { + const raw = new TypeError('fetch failed', { + cause: Object.assign(new Error('Headers Timeout Error'), { + name: 'HeadersTimeoutError', + code: 'UND_ERR_HEADERS_TIMEOUT', + }), + }); + const { err } = await callWith(throwing(raw)); + expect(err).toMatchObject({ + statusCode: 504, + legacyCode: 'upstream_timeout', + fields: { upstreamCode: 'UND_ERR_HEADERS_TIMEOUT' }, + }); + }); + it('passes an error with a sub-400 status through untranslated', async () => { const raw = { status: 302, message: 'redirected' }; const { err } = await callWith(throwing(raw)); @@ -435,3 +462,53 @@ describe('DriverController per-method rate limiting', () => { expect(alarms).toEqual([]); }); }); + +// -- Client disconnect -------------------------------------------------- + +describe('DriverController client disconnect', () => { + it('exposes an abort signal in the request context that fires when the client leaves early', async () => { + let seen: AbortSignal | undefined; + const { handler } = build({ + run: async () => { + seen = Context.get('abortSignal'); + await new Promise((r) => setImmediate(r)); + return { ok: true }; + }, + }); + const res = new MockRes(); + const done = runWithContext({}, () => + handler( + makeReq({ interface: 'test-iface', method: 'run' }), + res as unknown as Response, + () => {}, + ), + ); + res.destroy(); + await done; + + expect(seen).toBeInstanceOf(AbortSignal); + expect(seen?.aborted).toBe(true); + }); + + it('does not abort when the response simply finished', async () => { + let seen: AbortSignal | undefined; + const { handler } = build({ + run: () => { + seen = Context.get('abortSignal'); + return { ok: true }; + }, + }); + const res = new MockRes(); + await runWithContext({}, () => + handler( + makeReq({ interface: 'test-iface', method: 'run' }), + res as unknown as Response, + () => {}, + ), + ); + res.end(); + await new Promise((r) => setImmediate(r)); + + expect(seen?.aborted).toBe(false); + }); +}); diff --git a/src/backend/controllers/drivers/DriverController.ts b/src/backend/controllers/drivers/DriverController.ts index 5d11d0cb9..be3dba5c2 100644 --- a/src/backend/controllers/drivers/DriverController.ts +++ b/src/backend/controllers/drivers/DriverController.ts @@ -31,6 +31,7 @@ import { } from '../../core/http/middleware/rateLimit.js'; import type { PuterRouter } from '../../core/http/PuterRouter.js'; import type { DriverMeta } from '../../drivers/meta.js'; +import { isUpstreamTimeoutError } from '../../drivers/util/upstreamErrors.js'; import { isDriverStreamResult, resolveCallableMethods, @@ -116,9 +117,20 @@ const translateProviderError = (err: unknown): unknown => { message?: string; error?: { code?: string; type?: string; message?: string }; code?: string; + cause?: unknown; }; const status = extractUpstreamStatus(e); - if (typeof status !== 'number') return err; + if (typeof status !== 'number') { + if (isUpstreamTimeoutError(e)) { + const cause = e.cause as { code?: string } | undefined; + return new HttpError(504, 'AI provider timed out', { + legacyCode: 'upstream_timeout', + fields: { upstreamCode: e.code ?? cause?.code }, + cause: err, + }); + } + return err; + } const msg = e.error?.message ?? e.message ?? 'Upstream provider error'; const upstreamCode = e.error?.code ?? e.code; @@ -301,8 +313,7 @@ export class DriverController extends PuterController { if (req.actor) { const permService = this.services.permission as unknown as - | PermissionService - | undefined; + PermissionService | undefined; if (permService) { // Build via PermissionUtil.join so any `:` in a driver or // interface name is escaped — raw interpolation would let a @@ -434,6 +445,15 @@ export class DriverController extends PuterController { // a stale value from a prior call. Context.set('driverName', requestedDriver); + // A caller that hangs up mid-call gets nothing back, so long-running + // drivers watch this to stop working (and metering) as soon as it does. + // `close` after `finish` is the normal end of a response, not an abort. + const abort = new AbortController(); + res.once('close', () => { + if (!res.writableFinished) abort.abort(); + }); + Context.set('abortSignal', abort.signal); + // Per-method lifecycle events, scoped to `driver..`. // Subscribers can listen on `driver.*`, `driver..*`, or the // exact key. `before` is emitted via `emitAndWait` so a listener may diff --git a/src/backend/core/context.ts b/src/backend/core/context.ts index d775190dc..26ceb3160 100644 --- a/src/backend/core/context.ts +++ b/src/backend/core/context.ts @@ -61,6 +61,12 @@ export interface KnownContextFields { * `/drivers/call` dispatch); drivers read it to pick a provider. */ driverName: string; + /** + * Aborts when the client disconnects before the response has finished (set + * by DriverController). Long-running drivers poll it so work nobody will + * receive stops early and is never metered. + */ + abortSignal: AbortSignal; } // -- Context store --------------------------------------------------- diff --git a/src/backend/drivers/ai-chat/ChatCompletionDriver.routing.test.ts b/src/backend/drivers/ai-chat/ChatCompletionDriver.routing.test.ts index 95eb247fc..4d23b785f 100644 --- a/src/backend/drivers/ai-chat/ChatCompletionDriver.routing.test.ts +++ b/src/backend/drivers/ai-chat/ChatCompletionDriver.routing.test.ts @@ -419,3 +419,51 @@ describe('ChatCompletionDriver unhealthy-route skipping', () => { expect(attempts[0]).toMatchObject({ provider: 'infron' }); }); }); + +// A transport timeout carries no status, so the classifier used to lump a +// chain of them in with "our bug" and page. It is the provider's pace. +describe('ChatCompletionDriver timeout classification across the chain', () => { + // Shaped like the Stainless SDKs' timeout: no status, only the class. + class APIConnectionTimeoutError extends Error { + constructor() { + super('Request timed out.'); + } + } + + const completeShared = () => + withTestActor(() => + driver.complete({ + model: 'deepseek-v4-pro', + messages: [{ role: 'user', content: 'hi' }], + }), + ).catch((e: unknown) => e as HttpError); + + it('returns 504 upstream_timeout when every route timed out', async () => { + createMock.mockRejectedValue(new APIConnectionTimeoutError()); + + const err = await completeShared(); + + expect(err).toMatchObject({ + statusCode: 504, + legacyCode: 'upstream_timeout', + }); + const attempts = ( + err as unknown as { fields: { attempts: { timedOut?: boolean }[] } } + ).fields.attempts; + expect(attempts.length).toBeGreaterThan(1); + expect(attempts.every((a) => a.timedOut === true)).toBe(true); + }); + + it('counts a timed-out route as an upstream signal, so a mixed chain is not a 500', async () => { + createMock + .mockRejectedValueOnce(new APIConnectionTimeoutError()) + .mockRejectedValue(new Error('upstream down')); + + const err = await completeShared(); + + expect(err).toMatchObject({ + statusCode: 400, + legacyCode: 'upstream_failed', + }); + }); +}); diff --git a/src/backend/drivers/ai-chat/ChatCompletionDriver.test.ts b/src/backend/drivers/ai-chat/ChatCompletionDriver.test.ts index 48259d926..da58e631e 100644 --- a/src/backend/drivers/ai-chat/ChatCompletionDriver.test.ts +++ b/src/backend/drivers/ai-chat/ChatCompletionDriver.test.ts @@ -1118,6 +1118,36 @@ describe('ChatCompletionDriver.complete OpenAI-shape normalization', () => { // ── Fallback / error envelope ─────────────────────────────────────── describe('ChatCompletionDriver.complete fallback and error envelope', () => { + it('returns HTTP 504 upstream_timeout when the only route timed out', async () => { + // Shaped like the Stainless SDKs' timeout: no status, only the class. + class APIConnectionTimeoutError extends Error { + constructor() { + super('Request timed out.'); + } + } + vi.spyOn(FakeChatProvider.prototype, 'complete').mockRejectedValue( + new APIConnectionTimeoutError(), + ); + + const caught = await withTestActor(() => + driver.complete({ + model: 'fake', + messages: [{ role: 'user', content: 'hi' }], + }), + ).catch((e: unknown) => e as HttpError); + + expect(caught).toBeInstanceOf(HttpError); + expect(caught).toMatchObject({ + statusCode: 504, + legacyCode: 'upstream_timeout', + message: 'AI provider timed out', + }); + const attempts = (caught as unknown as { fields: { attempts: { timedOut?: boolean }[] } }) + .fields.attempts; + expect(attempts).toHaveLength(1); + expect(attempts[0].timedOut).toBe(true); + }); + it('returns HTTP 500 with the failure history in `fields.attempts` when all providers fail', async () => { vi.spyOn(FakeChatProvider.prototype, 'complete').mockRejectedValue( new Error('boom'), diff --git a/src/backend/drivers/ai-chat/ChatCompletionDriver.ts b/src/backend/drivers/ai-chat/ChatCompletionDriver.ts index 36e460c8b..af001868f 100644 --- a/src/backend/drivers/ai-chat/ChatCompletionDriver.ts +++ b/src/backend/drivers/ai-chat/ChatCompletionDriver.ts @@ -32,6 +32,7 @@ import { NO_CREDIT_HOLD } from '../../services/metering/types.js'; import type { DriverStreamResult } from '../meta.js'; import { PuterDriver } from '../types.js'; import { AI_CONCURRENT, AI_RATE_LIMIT } from '../util/aiLimits.js'; +import { isUpstreamTimeoutError } from '../util/upstreamErrors.js'; import { AlibabaProvider } from './providers/alibaba/AlibabaProvider.js'; import { AzureChatProvider } from './providers/azure/AzureChatProvider.js'; import { AzureResponsesProvider } from './providers/azure/AzureResponsesProvider.js'; @@ -111,6 +112,8 @@ type ProviderAttempt = { status?: number; code?: string; error: string; + /** The attempt died to a transport timeout rather than an answer. */ + timedOut?: boolean; }; /** @@ -146,6 +149,7 @@ const toAttempt = ( status, code: e?.error?.code ?? e?.code, error: message, + ...(isUpstreamTimeoutError(err) ? { timedOut: true } : {}), }; }; @@ -193,6 +197,7 @@ const routeId = (provider: string, modelId: string) => `${provider}:${modelId}`; * - All auth failures → 500 `upstream_auth_failed` (paged: our config) * - All upstream 5xx → 400 `upstream_provider_unavailable` (no page) * - All upstream 4xx (other) → 400 `upstream_bad_request` (no page) + * - All timed out → 504 `upstream_timeout` (no page) * - Mixed → 400 `upstream_failed` (no page) */ const classifyAttempts = ( @@ -241,10 +246,18 @@ const classifyAttempts = ( }); } + if (attempts.every((a) => a.timedOut)) { + return new HttpError(504, 'AI provider timed out', { + legacyCode: 'upstream_timeout', + fields, + }); + } + // Mixed failures where at least one attempt is clearly upstream - // (had an HTTP status from the SDK) means "AI providers couldn't - // satisfy the request" — expose, don't page. + // (had an HTTP status from the SDK, or never got one in time) means + // "AI providers couldn't satisfy the request" — expose, don't page. const isUpstreamSignal = (a: ProviderAttempt) => + a.timedOut === true || a.status !== undefined || isRateLimit(a) || isAuthFailure(a) || diff --git a/src/backend/drivers/ai-chat/providers/together/TogetherAIProvider.test.ts b/src/backend/drivers/ai-chat/providers/together/TogetherAIProvider.test.ts index 2c3393f62..ac62960cc 100644 --- a/src/backend/drivers/ai-chat/providers/together/TogetherAIProvider.test.ts +++ b/src/backend/drivers/ai-chat/providers/together/TogetherAIProvider.test.ts @@ -176,7 +176,10 @@ describe('TogetherAIProvider construction', () => { it('constructs the Together SDK with the configured API key', () => { makeProvider(); expect(togetherCtor).toHaveBeenCalledTimes(1); - expect(togetherCtor).toHaveBeenCalledWith({ apiKey: 'test-key' }); + expect(togetherCtor).toHaveBeenCalledWith({ + apiKey: 'test-key', + timeout: 600_000, + }); }); }); diff --git a/src/backend/drivers/ai-chat/providers/together/TogetherAIProvider.ts b/src/backend/drivers/ai-chat/providers/together/TogetherAIProvider.ts index 6bbd45337..968c13eff 100644 --- a/src/backend/drivers/ai-chat/providers/together/TogetherAIProvider.ts +++ b/src/backend/drivers/ai-chat/providers/together/TogetherAIProvider.ts @@ -42,8 +42,11 @@ export class TogetherAIProvider implements IChatProvider { #kvKey = 'togetherai:models'; constructor(config: { apiKey: string }, meteringService: MeteringService) { + // The SDK default is one minute, which long non-streaming + // completions exceed; match the ten minutes the other providers get. this.#together = new Together({ apiKey: config.apiKey, + timeout: 10 * 60 * 1000, }); this.#meteringService = meteringService; } diff --git a/src/backend/drivers/ai-image/providers/replicate/ReplicateImageGenerationProvider.ts b/src/backend/drivers/ai-image/providers/replicate/ReplicateImageGenerationProvider.ts index 5114f80fa..b7bd4f52f 100644 --- a/src/backend/drivers/ai-image/providers/replicate/ReplicateImageGenerationProvider.ts +++ b/src/backend/drivers/ai-image/providers/replicate/ReplicateImageGenerationProvider.ts @@ -29,30 +29,15 @@ import { REPLICATE_IMAGE_GENERATION_MODELS, type ReplicateImageModel, } from './models.js'; +import { + CONTENT_FILTER_PATTERN, + sanitizeUpstreamMessage, +} from '../../../util/upstreamErrors.js'; const DEFAULT_MODEL = 'black-forest-labs/flux-schnell'; const DEFAULT_RATIO = { w: 1024, h: 1024 }; const PREDICTION_FAILED_PREFIX = 'Prediction failed:'; -const MAX_UPSTREAM_MESSAGE_LENGTH = 300; -// Model-side content filters, as worded in failed-prediction errors. -const CONTENT_FILTER_PATTERN = - /\bnsfw\b|flagged as sensitive|sensitive content|content policy|\bE005\b/i; - -/** - * Strips markup and bounds length so an upstream HTML error page never rides - * through into a response body or an alarm signature. - */ -const sanitizeUpstreamMessage = (raw: string): string => { - const text = raw - .replace(/<(style|script)[\s\S]*?<\/\1>/gi, ' ') - .replace(/<[^>]*>/g, ' ') - .replace(/\s+/g, ' ') - .trim(); - return text.length > MAX_UPSTREAM_MESSAGE_LENGTH - ? `${text.slice(0, MAX_UPSTREAM_MESSAGE_LENGTH - 3)}...` - : text; -}; export class ReplicateImageGenerationProvider implements IImageProvider { static readonly #CORE_PARAMS: readonly string[] = [ diff --git a/src/backend/drivers/ai-video/providers/byteplus/BytePlusVideoProvider.test.ts b/src/backend/drivers/ai-video/providers/byteplus/BytePlusVideoProvider.test.ts index a3b2dc38d..394046838 100644 --- a/src/backend/drivers/ai-video/providers/byteplus/BytePlusVideoProvider.test.ts +++ b/src/backend/drivers/ai-video/providers/byteplus/BytePlusVideoProvider.test.ts @@ -44,6 +44,7 @@ import { PuterServer } from '../../../../server.js'; import { setupTestServer } from '../../../../testUtil.js'; import { withTestActor } from '../../../integrationTestUtil.js'; import { BYTEPLUS_VIDEO_GENERATION_MODELS } from './models.js'; +import { VIDEO_POLL_WINDOW_MS } from '../polling.js'; import { BytePlusVideoProvider } from './BytePlusVideoProvider.js'; // -- Test harness ---------------------------------------------------- @@ -378,6 +379,42 @@ describe('BytePlusVideoProvider.generate polling and outcomes', () => { expect(fetchSpy).toHaveBeenCalledTimes(4); // 1 create + 3 polls }); + it('gives up after the wait window as HttpError 504 upstream_timeout, without metering', async () => { + vi.useFakeTimers(); + try { + fetchSpy + .mockResolvedValueOnce( + jsonResponse({ id: 'cgt-slow', status: 'queued' }), + ) + .mockImplementation(async () => + jsonResponse({ id: 'cgt-slow', status: 'running' }), + ); + // The default poll interval keeps the window to ~60 polls. + const provider = new BytePlusVideoProvider( + { apiKey: 'test-key' }, + server.services.metering, + ); + + const rejection = withTestActor(() => + provider.generate({ prompt: 'hi' }), + ).catch((e: unknown) => e); + + // Ten-minute wait window, polled every 5s. + await vi.advanceTimersByTimeAsync(VIDEO_POLL_WINDOW_MS + 5_000); + + expect(await rejection).toMatchObject({ + statusCode: 504, + legacyCode: 'upstream_timeout', + message: + 'Timed out waiting for BytePlus video generation to complete', + fields: { provider: 'byteplus' }, + }); + expect(incrementUsageSpy).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + it('throws 400 upstream_failed when the task fails, without metering', async () => { mockTaskFlow({ id: 'cgt-test-1', @@ -386,7 +423,28 @@ describe('BytePlusVideoProvider.generate polling and outcomes', () => { }); await expect( withTestActor(() => makeProvider().generate({ prompt: 'hi' })), - ).rejects.toMatchObject({ statusCode: 400, message: 'blocked' }); + ).rejects.toMatchObject({ + statusCode: 400, + code: 'moderation_flagged', + message: 'blocked', + }); + expect(incrementUsageSpy).not.toHaveBeenCalled(); + }); + + it('throws 502 upstream_failed when the task fails on the provider side', async () => { + mockTaskFlow({ + id: 'cgt-test-1', + status: 'failed', + error: { code: 'InternalServiceError', message: 'worker crashed' }, + }); + await expect( + withTestActor(() => makeProvider().generate({ prompt: 'hi' })), + ).rejects.toMatchObject({ + statusCode: 502, + legacyCode: 'upstream_failed', + message: 'worker crashed', + fields: { provider: 'byteplus', upstreamCode: 'InternalServiceError' }, + }); expect(incrementUsageSpy).not.toHaveBeenCalled(); }); diff --git a/src/backend/drivers/ai-video/providers/byteplus/BytePlusVideoProvider.ts b/src/backend/drivers/ai-video/providers/byteplus/BytePlusVideoProvider.ts index ff0cac016..61eca6102 100644 --- a/src/backend/drivers/ai-video/providers/byteplus/BytePlusVideoProvider.ts +++ b/src/backend/drivers/ai-video/providers/byteplus/BytePlusVideoProvider.ts @@ -23,6 +23,7 @@ import type { MeteringService } from '../../../../services/metering/MeteringServ import type { IGenerateVideoParams, IVideoModel } from '../../types.js'; import { capSecondsToRemainingCredits } from '../../creditCap.js'; import { VideoProvider } from '../VideoProvider.js'; +import { pollUntilSettled, videoJobFailure } from '../polling.js'; import { BYTEPLUS_VIDEO_GENERATION_MODELS, BYTEPLUS_VIDEO_SPECS, @@ -32,7 +33,6 @@ import { const DEFAULT_TEST_VIDEO_URL = 'https://assets.puter.site/txt2vid.mp4'; const DEFAULT_BASE_URL = 'https://ark.ap-southeast.bytepluses.com/api/v3'; const DEFAULT_POLL_INTERVAL_MS = 5_000; -const DEFAULT_TIMEOUT_MS = 5 * 60 * 1000; const DEFAULT_MODEL = 'dreamina-seedance-2-0-mini-260615'; // Seedance 2.0 multimodal reference accepts up to 9 reference images. const MAX_REFERENCE_IMAGES = 9; @@ -202,14 +202,11 @@ export class BytePlusVideoProvider extends VideoProvider { const errorMessage = finalTask.error?.message ?? `Video generation ${finalTask.status}`; - // Ark's `failed` covers both user-input issues (content - // moderation) and upstream outages — same ambiguity as the - // Together provider, so expose it the same way: a 400 with - // `upstream_failed` that the alarm gate skips. - throw new HttpError(400, errorMessage, { - legacyCode: 'upstream_failed', - fields: { provider: 'byteplus' }, - }); + throw videoJobFailure( + 'byteplus', + errorMessage, + finalTask.error?.code, + ); } const videoUrl = finalTask.content?.video_url; @@ -336,22 +333,18 @@ export class BytePlusVideoProvider extends VideoProvider { } async #pollUntilComplete(taskId: string): Promise { - const start = Date.now(); - for (;;) { - const task = (await this.#request( - 'GET', - `/contents/generations/tasks/${taskId}`, - )) as ArkVideoTask; - if (task.status !== 'queued' && task.status !== 'running') { - return task; - } - if (Date.now() - start > DEFAULT_TIMEOUT_MS) { - throw new Error( - 'Timed out waiting for BytePlus video generation to complete', - ); - } - await this.#delay(this.#pollIntervalMs); - } + return await pollUntilSettled({ + provider: 'byteplus', + providerLabel: 'BytePlus', + intervalMs: this.#pollIntervalMs, + fetch: () => + this.#request( + 'GET', + `/contents/generations/tasks/${taskId}`, + ) as Promise, + isPending: (task) => + task.status === 'queued' || task.status === 'running', + }); } async #request( @@ -384,10 +377,6 @@ export class BytePlusVideoProvider extends VideoProvider { return payload; } - async #delay(ms: number): Promise { - return await new Promise((resolve) => setTimeout(resolve, ms)); - } - #getModel(requestedModel?: string): IVideoModel { const wanted = (requestedModel ?? '').trim().toLowerCase(); const found = BYTEPLUS_VIDEO_GENERATION_MODELS.find( diff --git a/src/backend/drivers/ai-video/providers/gemini/GeminiVideoProvider.test.ts b/src/backend/drivers/ai-video/providers/gemini/GeminiVideoProvider.test.ts index 73cb965dd..4a7ede645 100644 --- a/src/backend/drivers/ai-video/providers/gemini/GeminiVideoProvider.test.ts +++ b/src/backend/drivers/ai-video/providers/gemini/GeminiVideoProvider.test.ts @@ -45,6 +45,7 @@ import type { MeteringService } from '../../../../services/metering/MeteringServ import { PuterServer } from '../../../../server.js'; import { setupTestServer } from '../../../../testUtil.js'; import { withTestActor } from '../../../integrationTestUtil.js'; +import { VIDEO_POLL_WINDOW_MS } from '../polling.js'; import { GeminiVideoProvider } from './GeminiVideoProvider.js'; import { GEMINI_VIDEO_GENERATION_MODELS } from './models.js'; @@ -422,11 +423,72 @@ describe('GeminiVideoProvider.generate polling', () => { } }); - it('throws when the operation finishes with an error', async () => { + it('treats a failed poll as a missed poll rather than a failed job', async () => { + vi.useFakeTimers(); + try { + const provider = makeProvider(); + generateVideosMock.mockResolvedValueOnce({ done: false }); + getVideosOperationMock + .mockRejectedValueOnce( + Object.assign(new Error('unavailable'), { status: 503 }), + ) + .mockResolvedValueOnce(completedOperation()); + + const promise = withTestActor(() => + provider.generate({ + prompt: 'hi', + model: 'veo-3.1-generate-preview', + }), + ); + await vi.advanceTimersByTimeAsync(10_000); + await vi.advanceTimersByTimeAsync(10_000); + + expect(await promise).toBe('https://gemini/out.mp4'); + expect(getVideosOperationMock).toHaveBeenCalledTimes(2); + } finally { + vi.useRealTimers(); + } + }); + + it('gives up after the wait window as HttpError 504 upstream_timeout, without metering', async () => { + vi.useFakeTimers(); + try { + const provider = makeProvider(); + generateVideosMock.mockResolvedValueOnce({ done: false }); + getVideosOperationMock.mockResolvedValue({ done: false }); + + const rejection = withTestActor(() => + provider.generate({ + prompt: 'hi', + model: 'veo-3.1-generate-preview', + }), + ).catch((e: unknown) => e); + + // Ten-minute wait window, polled every 10s. + await vi.advanceTimersByTimeAsync(VIDEO_POLL_WINDOW_MS + 10_000); + + expect(await rejection).toMatchObject({ + statusCode: 504, + legacyCode: 'upstream_timeout', + message: + 'Timed out waiting for Gemini video generation to complete', + fields: { provider: 'gemini' }, + }); + expect(incrementUsageSpy).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it('surfaces an operation that finishes with an error as HttpError 502 upstream_failed', async () => { const provider = makeProvider(); generateVideosMock.mockResolvedValueOnce({ done: true, - error: { message: 'rate limit' }, + error: { + code: 13, + message: 'internal server issue', + status: 'INTERNAL', + }, response: {}, }); @@ -437,7 +499,12 @@ describe('GeminiVideoProvider.generate polling', () => { model: 'veo-3.1-generate-preview', }), ), - ).rejects.toThrow(/rate limit/); + ).rejects.toMatchObject({ + statusCode: 502, + legacyCode: 'upstream_failed', + message: 'internal server issue', + fields: { provider: 'gemini', upstreamCode: 'INTERNAL' }, + }); }); it('throws 400 with the filter reason when raiMediaFilteredCount > 0', async () => { @@ -460,7 +527,10 @@ describe('GeminiVideoProvider.generate polling', () => { ), ).rejects.toMatchObject({ statusCode: 400, + legacyCode: 'disallowed_value', + code: 'moderation_flagged', message: expect.stringContaining('unsafe content'), + fields: { provider: 'gemini' }, }); }); diff --git a/src/backend/drivers/ai-video/providers/gemini/GeminiVideoProvider.ts b/src/backend/drivers/ai-video/providers/gemini/GeminiVideoProvider.ts index eb4a5839d..55deb72ed 100644 --- a/src/backend/drivers/ai-video/providers/gemini/GeminiVideoProvider.ts +++ b/src/backend/drivers/ai-video/providers/gemini/GeminiVideoProvider.ts @@ -29,11 +29,11 @@ import type { MeteringService } from '../../../../services/metering/MeteringServ import type { IGenerateVideoParams, IVideoModel } from '../../types.js'; import { capSecondsToRemainingCredits } from '../../creditCap.js'; import { VideoProvider } from '../VideoProvider.js'; +import { pollUntilSettled, videoJobFailure } from '../polling.js'; import { GEMINI_VIDEO_GENERATION_MODELS, IGeminiVideoModel } from './models.js'; const DEFAULT_TEST_VIDEO_URL = 'https://assets.puter.site/txt2vid.mp4'; const POLL_INTERVAL_MS = 10_000; -const DEFAULT_TIMEOUT_MS = 10 * 60 * 1000; const DIMENSION_MAP: Record< string, @@ -230,7 +230,11 @@ export class GeminiVideoProvider extends VideoProvider { throw new HttpError( 400, `Video was filtered due to ${reasons}`, - { legacyCode: 'disallowed_value' }, + { + legacyCode: 'disallowed_value', + code: 'moderation_flagged', + fields: { provider: 'gemini' }, + }, ); } throw new Error('Gemini response did not include a video'); @@ -271,27 +275,28 @@ export class GeminiVideoProvider extends VideoProvider { async #pollUntilComplete( operation: GenerateVideosOperation, ): Promise { - let op = operation; - const start = Date.now(); - - while (!op.done) { - if (Date.now() - start > DEFAULT_TIMEOUT_MS) { - throw new Error( - 'Timed out waiting for Gemini video generation to complete', - ); - } - - await this.#delay(POLL_INTERVAL_MS); - op = await this.#client.operations.getVideosOperation({ - operation: op, - }); - } + const op = await pollUntilSettled({ + provider: 'gemini', + providerLabel: 'Gemini', + intervalMs: POLL_INTERVAL_MS, + initial: operation, + fetch: (previous) => + this.#client.operations.getVideosOperation({ + operation: previous ?? operation, + }), + isPending: (o) => !o.done, + }); if (op.error) { const msg = - (op.error as Record).message ?? - JSON.stringify(op.error); - throw new Error(`Gemini video generation failed: ${msg}`); + typeof op.error.message === 'string' + ? op.error.message + : JSON.stringify(op.error); + const code = + typeof op.error.status === 'string' + ? op.error.status + : undefined; + throw videoJobFailure('gemini', msg, code); } return op; @@ -354,8 +359,4 @@ export class GeminiVideoProvider extends VideoProvider { } return undefined; } - - async #delay(ms: number): Promise { - return await new Promise((resolve) => setTimeout(resolve, ms)); - } } diff --git a/src/backend/drivers/ai-video/providers/openai/OpenAIVideoProvider.test.ts b/src/backend/drivers/ai-video/providers/openai/OpenAIVideoProvider.test.ts index 89caa4ad3..2872fd431 100644 --- a/src/backend/drivers/ai-video/providers/openai/OpenAIVideoProvider.test.ts +++ b/src/backend/drivers/ai-video/providers/openai/OpenAIVideoProvider.test.ts @@ -448,6 +448,42 @@ describe('OpenAIVideoProvider.generate polling', () => { } }); + it('gives up after the wait window as HttpError 504 upstream_timeout, without metering', async () => { + vi.useFakeTimers(); + try { + const provider = makeProvider(); + videosCreateMock.mockResolvedValueOnce({ + id: 'job-slow', + status: 'queued', + size: '720x1280', + seconds: '4', + }); + videosRetrieveMock.mockResolvedValue({ + id: 'job-slow', + status: 'in_progress', + }); + + const rejection = withTestActor(() => + provider.generate({ prompt: 'hi', model: 'sora-2' }), + ).catch((e: unknown) => e); + + // Five-minute wait window, polled every 5s. + await vi.advanceTimersByTimeAsync(5 * 60 * 1000 + 5_000); + + expect(await rejection).toMatchObject({ + statusCode: 504, + legacyCode: 'upstream_timeout', + message: + 'Timed out waiting for Sora video generation to complete', + fields: { provider: 'openai' }, + }); + expect(videosDownloadContentMock).not.toHaveBeenCalled(); + expect(incrementUsageSpy).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + it('surfaces failed jobs as HttpError 400 upstream_failed (not a 500 page)', async () => { const provider = makeProvider(); videosCreateMock.mockResolvedValueOnce({ diff --git a/src/backend/drivers/ai-video/providers/openai/OpenAIVideoProvider.ts b/src/backend/drivers/ai-video/providers/openai/OpenAIVideoProvider.ts index 11e0bf2e2..47f59f4d0 100644 --- a/src/backend/drivers/ai-video/providers/openai/OpenAIVideoProvider.ts +++ b/src/backend/drivers/ai-video/providers/openai/OpenAIVideoProvider.ts @@ -25,6 +25,7 @@ import type { MeteringService } from '../../../../services/metering/MeteringServ import type { IGenerateVideoParams, IVideoModel } from '../../types.js'; import { capSecondsToRemainingCredits } from '../../creditCap.js'; import { VideoProvider } from '../VideoProvider.js'; +import { pollTimeoutError } from '../polling.js'; import { OPENAI_VIDEO_MODELS, OPENAI_VIDEO_ALLOWED_SECONDS } from './models.js'; import { Readable } from 'stream'; @@ -216,9 +217,7 @@ export class OpenAIVideoProvider extends VideoProvider { while (job.status === 'queued' || job.status === 'in_progress') { if (Date.now() - start > DEFAULT_TIMEOUT_MS) { - throw new Error( - 'Timed out waiting for Sora video generation to complete', - ); + throw pollTimeoutError('openai', 'Sora'); } await this.#delay(POLL_INTERVAL_MS); diff --git a/src/backend/drivers/ai-video/providers/polling.test.ts b/src/backend/drivers/ai-video/providers/polling.test.ts new file mode 100644 index 000000000..abef4a062 --- /dev/null +++ b/src/backend/drivers/ai-video/providers/polling.test.ts @@ -0,0 +1,297 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { Context, runWithContext } from '../../../core/context.js'; +import { HttpError } from '../../../core/http/HttpError.js'; +import { + VIDEO_POLL_WINDOW_MS, + abortableDelay, + clientAbortedError, + pollTimeoutError, + pollUntilSettled, + videoJobFailure, +} from './polling.js'; + +afterEach(() => { + vi.useRealTimers(); +}); + +describe('pollTimeoutError', () => { + it('builds a 504 upstream_timeout that names the provider and keeps the cause', () => { + const cause = new Error('last poll failed'); + const err = pollTimeoutError('together', 'Together AI', cause); + expect(err).toBeInstanceOf(HttpError); + expect(err).toMatchObject({ + statusCode: 504, + legacyCode: 'upstream_timeout', + message: + 'Timed out waiting for Together AI video generation to complete', + fields: { provider: 'together' }, + cause, + }); + }); +}); + +describe('clientAbortedError', () => { + it('builds a 400 client_aborted that names the provider', () => { + expect(clientAbortedError('gemini')).toMatchObject({ + statusCode: 400, + legacyCode: 'client_aborted', + fields: { provider: 'gemini' }, + }); + }); +}); + +describe('videoJobFailure', () => { + it('maps content-filter wording to 400 moderation_flagged', () => { + expect( + videoJobFailure('together', 'content policy violation'), + ).toMatchObject({ + statusCode: 400, + legacyCode: 'bad_request', + code: 'moderation_flagged', + message: 'content policy violation', + fields: { provider: 'together' }, + }); + }); + + it('reads the filter from the code when the message does not say', () => { + expect( + videoJobFailure( + 'byteplus', + 'blocked', + 'OutputVideoSensitiveContentDetected', + ), + ).toMatchObject({ + statusCode: 400, + code: 'moderation_flagged', + fields: { + provider: 'byteplus', + upstreamCode: 'OutputVideoSensitiveContentDetected', + }, + }); + }); + + it('maps a rejected parameter to 400 upstream_bad_request', () => { + expect( + videoJobFailure( + 'together', + 'fps is not supported by this model', + 'unsupportedParameter', + ), + ).toMatchObject({ + statusCode: 400, + legacyCode: 'upstream_bad_request', + }); + }); + + it('reads Together-style camelCase input codes as rejections', () => { + for (const [code, message] of [ + ['invalidDuration', "Invalid value for 'seconds' parameter."], + [ + 'missingFrameImagesForImageToVideoModel', + 'Missing required parameter for Kling 2.1 Standard image-to-video model', + ], + ]) { + expect(videoJobFailure('together', message, code)).toMatchObject({ + statusCode: 400, + legacyCode: 'upstream_bad_request', + fields: { upstreamCode: code }, + }); + } + }); + + it('maps anything else to 502 upstream_failed and strips markup', () => { + const err = videoJobFailure( + 'gemini', + '

Bad gateway

', + ); + expect(err).toMatchObject({ + statusCode: 502, + legacyCode: 'upstream_failed', + message: 'Bad gateway', + }); + }); + + it('falls back to a generic message when the provider sent none', () => { + expect(videoJobFailure('gemini', '').message).toBe( + 'Video generation failed', + ); + }); +}); + +describe('abortableDelay', () => { + it('resolves early when the signal aborts', async () => { + vi.useFakeTimers(); + const abort = new AbortController(); + const delay = abortableDelay(60_000, abort.signal); + abort.abort(); + await expect(delay).resolves.toBeUndefined(); + }); + + it('resolves after the delay otherwise', async () => { + vi.useFakeTimers(); + let done = false; + void abortableDelay(1_000).then(() => { + done = true; + }); + await vi.advanceTimersByTimeAsync(999); + expect(done).toBe(false); + await vi.advanceTimersByTimeAsync(1); + expect(done).toBe(true); + }); +}); + +describe('pollUntilSettled', () => { + type Job = { status: 'pending' | 'done' }; + const pending: Job = { status: 'pending' }; + const done: Job = { status: 'done' }; + + const poll = ( + fetch: (previous: Job | undefined) => Promise, + extra: { initial?: Job; windowMs?: number } = {}, + ) => + runWithContext({}, () => + pollUntilSettled({ + provider: 'together', + providerLabel: 'Together AI', + intervalMs: 5_000, + fetch, + isPending: (job) => job.status === 'pending', + ...extra, + }), + ); + + it('fetches immediately, then on the interval, until the job settles', async () => { + vi.useFakeTimers(); + const fetch = vi + .fn<(p: Job | undefined) => Promise>() + .mockResolvedValueOnce(pending) + .mockResolvedValueOnce(pending) + .mockResolvedValueOnce(done); + + const result = poll(fetch); + await vi.advanceTimersByTimeAsync(0); + expect(fetch).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(5_000); + await vi.advanceTimersByTimeAsync(5_000); + + expect(await result).toBe(done); + expect(fetch).toHaveBeenCalledTimes(3); + }); + + it('waits one interval before re-fetching a pending initial state', async () => { + vi.useFakeTimers(); + const fetch = vi + .fn<(p: Job | undefined) => Promise>() + .mockResolvedValueOnce(done); + + const result = poll(fetch, { initial: pending }); + await vi.advanceTimersByTimeAsync(0); + expect(fetch).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(5_000); + + expect(await result).toBe(done); + expect(fetch).toHaveBeenCalledWith(pending); + }); + + it('treats a transient poll failure as a missed poll', async () => { + vi.useFakeTimers(); + const fetch = vi + .fn<(p: Job | undefined) => Promise>() + .mockRejectedValueOnce( + Object.assign(new Error('bad gateway'), { status: 502 }), + ) + .mockResolvedValueOnce(done); + + const result = poll(fetch); + await vi.advanceTimersByTimeAsync(5_000); + + expect(await result).toBe(done); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it('rethrows a poll failure that is the provider\'s verdict', async () => { + const notFound = Object.assign(new Error('no such job'), { + status: 404, + }); + await expect(poll(() => Promise.reject(notFound))).rejects.toBe( + notFound, + ); + }); + + it('gives up after the window with a 504 carrying the last poll error', async () => { + vi.useFakeTimers(); + const flaky = Object.assign(new Error('still down'), { status: 503 }); + const fetch = vi + .fn<(p: Job | undefined) => Promise>() + .mockRejectedValue(flaky); + + const rejection = poll(fetch).catch((e: unknown) => e); + await vi.advanceTimersByTimeAsync(VIDEO_POLL_WINDOW_MS + 5_000); + + expect(await rejection).toMatchObject({ + statusCode: 504, + legacyCode: 'upstream_timeout', + cause: flaky, + }); + }); + + it('honours a shorter window when one is given', async () => { + vi.useFakeTimers(); + const rejection = poll(() => Promise.resolve(pending), { + windowMs: 20_000, + }).catch((e: unknown) => e); + await vi.advanceTimersByTimeAsync(25_000); + + expect(await rejection).toMatchObject({ statusCode: 504 }); + }); + + it('stops as soon as the request context signals a client abort', async () => { + vi.useFakeTimers(); + const abort = new AbortController(); + const fetch = vi + .fn<(p: Job | undefined) => Promise>() + .mockResolvedValue(pending); + + const rejection = runWithContext({}, () => { + Context.set('abortSignal', abort.signal); + return pollUntilSettled({ + provider: 'gemini', + providerLabel: 'Gemini', + intervalMs: 10_000, + fetch, + isPending: (job) => job.status === 'pending', + }); + }).catch((e: unknown) => e); + + await vi.advanceTimersByTimeAsync(10_000); + abort.abort(); + await vi.advanceTimersByTimeAsync(0); + + expect(await rejection).toMatchObject({ + statusCode: 400, + legacyCode: 'client_aborted', + fields: { provider: 'gemini' }, + }); + // The abort cut the second wait short; no third poll went out. + expect(fetch).toHaveBeenCalledTimes(2); + }); +}); diff --git a/src/backend/drivers/ai-video/providers/polling.ts b/src/backend/drivers/ai-video/providers/polling.ts new file mode 100644 index 000000000..b0f747ce1 --- /dev/null +++ b/src/backend/drivers/ai-video/providers/polling.ts @@ -0,0 +1,163 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import { Context } from '../../../core/context.js'; +import { HttpError } from '../../../core/http/HttpError.js'; +import { + CONTENT_FILTER_PATTERN, + isTransientUpstreamError, + sanitizeUpstreamMessage, +} from '../../util/upstreamErrors.js'; + +/** How long a provider's job may run before we stop waiting for it. */ +export const VIDEO_POLL_WINDOW_MS = 10 * 60 * 1000; + +// Provider wording for input the model does not accept, including Together's +// camelCase codes (`invalidDuration`, `missingFrameImagesForImageToVideoModel`). +const INVALID_INPUT_PATTERN = + /\b(unsupported|invalid|missing)\w*|not supported|required parameter/i; + +/** + * The upstream job outlived the wait window. That is the provider's pace, not a + * fault of ours, so it leaves as a 504 tagged `upstream_timeout`, which the + * alarm gate skips. + */ +export const pollTimeoutError = ( + provider: string, + providerLabel: string, + cause?: unknown, +): HttpError => + new HttpError( + 504, + `Timed out waiting for ${providerLabel} video generation to complete`, + { + legacyCode: 'upstream_timeout', + fields: { provider }, + ...(cause !== undefined ? { cause } : {}), + }, + ); + +/** The caller hung up mid-job; nobody is left to bill or answer. */ +export const clientAbortedError = (provider: string): HttpError => + new HttpError(400, 'Client disconnected before video generation finished', { + legacyCode: 'client_aborted', + fields: { provider }, + }); + +/** + * Classifies a job the provider reports as failed. A content-filter refusal is + * the caller's to act on; a rejected parameter is theirs to fix; anything else + * failed on the provider's side. + */ +export const videoJobFailure = ( + provider: string, + message: string, + code?: string, +): HttpError => { + const detail = + sanitizeUpstreamMessage(message) || 'Video generation failed'; + const haystack = `${code ?? ''} ${detail}`; + const fields = { provider, upstreamCode: code }; + if (CONTENT_FILTER_PATTERN.test(haystack)) { + return new HttpError(400, detail, { + legacyCode: 'bad_request', + code: 'moderation_flagged', + fields, + }); + } + if (INVALID_INPUT_PATTERN.test(haystack)) { + return new HttpError(400, detail, { + legacyCode: 'upstream_bad_request', + fields, + }); + } + return new HttpError(502, detail, { + legacyCode: 'upstream_failed', + fields, + }); +}; + +/** Resolves after `ms`, or as soon as `signal` aborts. */ +export const abortableDelay = ( + ms: number, + signal?: AbortSignal, +): Promise => + new Promise((resolve) => { + if (signal?.aborted) return resolve(); + const done = () => { + clearTimeout(timer); + signal?.removeEventListener('abort', done); + resolve(); + }; + const timer = setTimeout(done, ms); + signal?.addEventListener('abort', done, { once: true }); + }); + +export interface PollOptions { + /** Goes into `fields.provider` on every error raised here. */ + provider: string; + /** Human-readable name for the timeout message. */ + providerLabel: string; + intervalMs: number; + windowMs?: number; + /** The state the create call already returned, if it returned one. */ + initial?: T; + /** Fetches the current job state; receives the last state seen. */ + fetch: (previous: T | undefined) => Promise; + isPending: (job: T) => boolean; +} + +/** + * Polls a provider job until it settles. A transient poll failure is a missed + * poll, not a failed job; a client disconnect ends the wait before anything is + * metered; the window closes with a 504 that carries the last poll error. + */ +export const pollUntilSettled = async (opts: PollOptions): Promise => { + const windowMs = opts.windowMs ?? VIDEO_POLL_WINDOW_MS; + const signal = Context.get('abortSignal'); + const start = Date.now(); + let job = opts.initial; + let lastError: unknown; + let firstFetch = job === undefined; + + for (;;) { + if (job !== undefined && !opts.isPending(job)) return job; + if (signal?.aborted) throw clientAbortedError(opts.provider); + if (Date.now() - start > windowMs) { + throw pollTimeoutError( + opts.provider, + opts.providerLabel, + lastError, + ); + } + if (!firstFetch) { + await abortableDelay(opts.intervalMs, signal); + // The delay ends early on abort; do not spend a poll on it. + if (signal?.aborted) throw clientAbortedError(opts.provider); + } + firstFetch = false; + try { + job = await opts.fetch(job); + lastError = undefined; + } catch (err) { + if (!isTransientUpstreamError(err)) throw err; + lastError = err; + } + } +}; diff --git a/src/backend/drivers/ai-video/providers/together/TogetherVideoProvider.test.ts b/src/backend/drivers/ai-video/providers/together/TogetherVideoProvider.test.ts index 33356acba..6dd70c09d 100644 --- a/src/backend/drivers/ai-video/providers/together/TogetherVideoProvider.test.ts +++ b/src/backend/drivers/ai-video/providers/together/TogetherVideoProvider.test.ts @@ -40,10 +40,12 @@ import { type MockInstance, } from 'vitest'; +import { Context } from '../../../../core/context.js'; import type { MeteringService } from '../../../../services/metering/MeteringService.js'; import { PuterServer } from '../../../../server.js'; import { setupTestServer } from '../../../../testUtil.js'; import { withTestActor } from '../../../integrationTestUtil.js'; +import { VIDEO_POLL_WINDOW_MS } from '../polling.js'; import { TogetherVideoProvider } from './TogetherVideoProvider.js'; import { TOGETHER_VIDEO_GENERATION_MODELS } from './models.js'; @@ -118,7 +120,10 @@ describe('TogetherVideoProvider construction', () => { it('constructs the Together SDK with the configured api key', () => { makeProvider(); expect(togetherCtor).toHaveBeenCalledTimes(1); - expect(togetherCtor).toHaveBeenCalledWith({ apiKey: 'test-key' }); + expect(togetherCtor).toHaveBeenCalledWith({ + apiKey: 'test-key', + timeout: 60_000, + }); }); it('throws when no apiKey is supplied', () => { @@ -409,7 +414,96 @@ describe('TogetherVideoProvider.generate polling', () => { } }); - it('surfaces failed jobs as HttpError 400 upstream_failed (not 500) so they do not page', async () => { + it('gives up after the wait window as HttpError 504 upstream_timeout, without metering', async () => { + vi.useFakeTimers(); + try { + const provider = makeProvider(); + videosCreateMock.mockResolvedValueOnce({ id: 'job-slow' }); + videosRetrieveMock.mockResolvedValue({ + id: 'job-slow', + status: 'in_progress', + }); + + const rejection = withTestActor(() => + provider.generate({ prompt: 'hi' }), + ).catch((e: unknown) => e); + + // Ten-minute wait window, polled every 5s. + await vi.advanceTimersByTimeAsync(VIDEO_POLL_WINDOW_MS + 5_000); + + expect(await rejection).toMatchObject({ + statusCode: 504, + legacyCode: 'upstream_timeout', + message: + 'Timed out waiting for Together AI video generation to complete', + fields: { provider: 'together' }, + }); + expect(incrementUsageSpy).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it('treats a failed poll as a missed poll rather than a failed job', async () => { + vi.useFakeTimers(); + try { + const provider = makeProvider(); + videosCreateMock.mockResolvedValueOnce({ id: 'job-flaky' }); + videosRetrieveMock + .mockRejectedValueOnce( + Object.assign(new Error('bad gateway'), { status: 502 }), + ) + .mockResolvedValueOnce({ + id: 'job-flaky', + status: 'completed', + outputs: { video_url: 'https://together/flaky.mp4' }, + }); + + const promise = withTestActor(() => + provider.generate({ prompt: 'hi' }), + ); + await vi.advanceTimersByTimeAsync(5_000); + + expect(await promise).toBe('https://together/flaky.mp4'); + expect(videosRetrieveMock).toHaveBeenCalledTimes(2); + } finally { + vi.useRealTimers(); + } + }); + + it('stops polling and skips metering when the client disconnects', async () => { + vi.useFakeTimers(); + try { + const provider = makeProvider(); + videosCreateMock.mockResolvedValueOnce({ id: 'job-gone' }); + videosRetrieveMock.mockResolvedValue({ + id: 'job-gone', + status: 'in_progress', + }); + const abort = new AbortController(); + + const rejection = withTestActor(() => { + Context.set('abortSignal', abort.signal); + return provider.generate({ prompt: 'hi' }); + }).catch((e: unknown) => e); + + await vi.advanceTimersByTimeAsync(5_000); + abort.abort(); + await vi.advanceTimersByTimeAsync(0); + + expect(await rejection).toMatchObject({ + statusCode: 400, + legacyCode: 'client_aborted', + fields: { provider: 'together' }, + }); + expect(videosRetrieveMock).toHaveBeenCalledTimes(2); + expect(incrementUsageSpy).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it('surfaces a content-policy refusal as 400 moderation_flagged', async () => { const provider = makeProvider(); videosCreateMock.mockResolvedValueOnce({ id: 'job-3' }); videosRetrieveMock.mockResolvedValueOnce({ @@ -422,9 +516,54 @@ describe('TogetherVideoProvider.generate polling', () => { withTestActor(() => provider.generate({ prompt: 'hi' })), ).rejects.toMatchObject({ statusCode: 400, - legacyCode: 'upstream_failed', + legacyCode: 'bad_request', + code: 'moderation_flagged', message: 'content policy violation', + fields: { provider: 'together' }, }); + expect(incrementUsageSpy).not.toHaveBeenCalled(); + }); + + it('surfaces a rejected parameter as 400 upstream_bad_request', async () => { + const provider = makeProvider(); + videosCreateMock.mockResolvedValueOnce({ id: 'job-4' }); + videosRetrieveMock.mockResolvedValueOnce({ + id: 'job-4', + status: 'failed', + error: { + code: 'unsupportedParameter', + message: 'fps is not supported by this model', + }, + }); + + await expect( + withTestActor(() => provider.generate({ prompt: 'hi' })), + ).rejects.toMatchObject({ + statusCode: 400, + legacyCode: 'upstream_bad_request', + message: 'fps is not supported by this model', + fields: { provider: 'together', upstreamCode: 'unsupportedParameter' }, + }); + }); + + it('surfaces any other failed job as 502 upstream_failed, without metering', async () => { + const provider = makeProvider(); + videosCreateMock.mockResolvedValueOnce({ id: 'job-5' }); + videosRetrieveMock.mockResolvedValueOnce({ + id: 'job-5', + status: 'failed', + error: { message: 'worker crashed' }, + }); + + await expect( + withTestActor(() => provider.generate({ prompt: 'hi' })), + ).rejects.toMatchObject({ + statusCode: 502, + legacyCode: 'upstream_failed', + message: 'worker crashed', + fields: { provider: 'together' }, + }); + expect(incrementUsageSpy).not.toHaveBeenCalled(); }); it('throws when a finished job has no video_url', async () => { diff --git a/src/backend/drivers/ai-video/providers/together/TogetherVideoProvider.ts b/src/backend/drivers/ai-video/providers/together/TogetherVideoProvider.ts index a5886d2fd..535b32667 100644 --- a/src/backend/drivers/ai-video/providers/together/TogetherVideoProvider.ts +++ b/src/backend/drivers/ai-video/providers/together/TogetherVideoProvider.ts @@ -23,11 +23,12 @@ import { HttpError } from '../../../../core/http/HttpError.js'; import type { MeteringService } from '../../../../services/metering/MeteringService.js'; import type { IGenerateVideoParams, IVideoModel } from '../../types.js'; import { VideoProvider } from '../VideoProvider.js'; +import { pollUntilSettled, videoJobFailure } from '../polling.js'; import { TOGETHER_VIDEO_GENERATION_MODELS } from './models.js'; const DEFAULT_TEST_VIDEO_URL = 'https://assets.puter.site/txt2vid.mp4'; const POLL_INTERVAL_MS = 5_000; -const DEFAULT_TIMEOUT_MS = 5 * 60 * 1000; +const REQUEST_TIMEOUT_MS = 60 * 1000; const DEFAULT_MODEL = 'minimax/video-01-director'; const DEFAULT_DURATION_SECONDS = 6; @@ -40,7 +41,12 @@ export class TogetherVideoProvider extends VideoProvider { if (!config.apiKey) { throw new Error('Together AI video generation requires an API key'); } - this.#client = new Together({ apiKey: config.apiKey }); + // Bounds each create/retrieve call; a slow poll is retried by the + // loop rather than failing the job. + this.#client = new Together({ + apiKey: config.apiKey, + timeout: REQUEST_TIMEOUT_MS, + }); this.#meteringService = meteringService; } @@ -190,24 +196,22 @@ export class TogetherVideoProvider extends VideoProvider { if (finalJob.status === 'failed') { const errorMessage = - finalJob?.error.message ?? + finalJob?.error?.message ?? finalJob?.info?.errors?.[0]?.message ?? finalJob?.info?.errors?.message ?? finalJob?.info?.errors ?? 'Video generation failed'; - // Together returns `failed` for both user-input issues - // (content policy / unsupported params) and their own - // outages — we can't reliably tell from the payload, so - // expose as 4xx with `upstream_failed`. The alarm gate - // skips `upstream_*` legacy codes so this no longer pages. - throw new HttpError(400, errorMessage, { - legacyCode: 'upstream_failed', - fields: { provider: 'together' }, - }); + throw videoJobFailure( + 'together', + typeof errorMessage === 'string' + ? errorMessage + : JSON.stringify(errorMessage), + finalJob?.error?.code, + ); } if (finalJob.status === 'cancelled') { - throw new Error('Video generation was cancelled'); + throw videoJobFailure('together', 'Video generation was cancelled'); } const usageKey = `together-video:${model}`; @@ -228,25 +232,14 @@ export class TogetherVideoProvider extends VideoProvider { async #pollUntilComplete(jobId: string): Promise { // any here because sdk types are wrong https://docs.together.ai/docs/videos-overview -> "Job Status Reference" - let job = await (this.#client as any).videos.retrieve(jobId); - const start = Date.now(); - - while (job.status === 'queued' || job.status === 'in_progress') { - if (Date.now() - start > DEFAULT_TIMEOUT_MS) { - throw new Error( - 'Timed out waiting for Together AI video generation to complete', - ); - } - - await this.#delay(POLL_INTERVAL_MS); - job = await (this.#client as any).videos.retrieve(jobId); - } - - return job; - } - - async #delay(ms: number): Promise { - return await new Promise((resolve) => setTimeout(resolve, ms)); + return await pollUntilSettled({ + provider: 'together', + providerLabel: 'Together AI', + intervalMs: POLL_INTERVAL_MS, + fetch: () => (this.#client as any).videos.retrieve(jobId), + isPending: (job) => + job.status === 'queued' || job.status === 'in_progress', + }); } async #getModel(requestedModel?: string): Promise { diff --git a/src/backend/drivers/util/upstreamErrors.test.ts b/src/backend/drivers/util/upstreamErrors.test.ts new file mode 100644 index 000000000..4c1ed8abc --- /dev/null +++ b/src/backend/drivers/util/upstreamErrors.test.ts @@ -0,0 +1,129 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import { APIConnectionError, APIConnectionTimeoutError } from 'openai'; +import { describe, expect, it } from 'vitest'; +import { HttpError } from '../../core/http/HttpError.js'; +import { + CONTENT_FILTER_PATTERN, + isTransientUpstreamError, + isUpstreamTimeoutError, + sanitizeUpstreamMessage, +} from './upstreamErrors.js'; + +const withStatus = (status: number) => + Object.assign(new Error(`status ${status}`), { status }); + +describe('isUpstreamTimeoutError', () => { + it('recognises the Stainless SDK timeout class', () => { + expect(isUpstreamTimeoutError(new APIConnectionTimeoutError())).toBe( + true, + ); + }); + + it('recognises an undici timeout wrapped in a fetch failed TypeError', () => { + const err = new TypeError('fetch failed', { + cause: Object.assign(new Error('Headers Timeout Error'), { + name: 'HeadersTimeoutError', + code: 'UND_ERR_HEADERS_TIMEOUT', + }), + }); + expect(isUpstreamTimeoutError(err)).toBe(true); + }); + + it('recognises an AbortSignal.timeout rejection by name', () => { + expect(isUpstreamTimeoutError({ name: 'TimeoutError' })).toBe(true); + }); + + it('ignores ordinary errors, non-objects, and status-bearing failures', () => { + expect(isUpstreamTimeoutError(new Error('boom'))).toBe(false); + expect(isUpstreamTimeoutError('timed out')).toBe(false); + expect(isUpstreamTimeoutError(withStatus(504))).toBe(false); + }); +}); + +describe('isTransientUpstreamError', () => { + it('treats timeouts and dropped connections as transient', () => { + expect(isTransientUpstreamError(new APIConnectionTimeoutError())).toBe( + true, + ); + expect( + isTransientUpstreamError(new APIConnectionError({ message: 'x' })), + ).toBe(true); + expect( + isTransientUpstreamError( + Object.assign(new Error('reset'), { code: 'ECONNRESET' }), + ), + ).toBe(true); + }); + + it('treats 5xx, 408 and 429 statuses as transient, however they are carried', () => { + expect(isTransientUpstreamError(withStatus(503))).toBe(true); + expect(isTransientUpstreamError(withStatus(429))).toBe(true); + expect(isTransientUpstreamError(withStatus(408))).toBe(true); + expect(isTransientUpstreamError(new HttpError(502, 'x'))).toBe(true); + expect( + isTransientUpstreamError({ response: { status: 500 } }), + ).toBe(true); + }); + + it('does not treat other 4xx responses or plain errors as transient', () => { + expect(isTransientUpstreamError(withStatus(404))).toBe(false); + expect(isTransientUpstreamError(withStatus(400))).toBe(false); + expect(isTransientUpstreamError(new HttpError(400, 'x'))).toBe(false); + expect(isTransientUpstreamError(new Error('boom'))).toBe(false); + }); +}); + +describe('CONTENT_FILTER_PATTERN', () => { + it('matches the wording providers use for refusals', () => { + for (const text of [ + 'Error generating image: NSFW content detected.', + 'OutputVideoSensitiveContentDetected', + 'content policy violation', + 'blocked by our safety filters', + 'The input or output was flagged as sensitive. (E005)', + ]) { + expect(text).toMatch(CONTENT_FILTER_PATTERN); + } + }); + + it('does not match ordinary failures', () => { + expect('q_descale must have shape (batch_size, num_heads_k)').not.toMatch( + CONTENT_FILTER_PATTERN, + ); + expect('internal server issue').not.toMatch(CONTENT_FILTER_PATTERN); + }); +}); + +describe('sanitizeUpstreamMessage', () => { + it('strips markup and collapses whitespace', () => { + expect( + sanitizeUpstreamMessage( + '

Bad

\n

gateway

', + ), + ).toBe('Bad gateway'); + }); + + it('bounds the length', () => { + const out = sanitizeUpstreamMessage('x'.repeat(1000)); + expect(out.length).toBe(300); + expect(out.endsWith('...')).toBe(true); + }); +}); diff --git a/src/backend/drivers/util/upstreamErrors.ts b/src/backend/drivers/util/upstreamErrors.ts new file mode 100644 index 000000000..482932ca2 --- /dev/null +++ b/src/backend/drivers/util/upstreamErrors.ts @@ -0,0 +1,129 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +// -- Transport failures -- + +// Transport timeouts carry no HTTP status. The Stainless SDKs (OpenAI, +// Together, Gemini) throw `APIConnectionTimeoutError`; undici wraps its own +// in a `fetch failed` TypeError whose cause carries the code. +const TIMEOUT_ERROR_NAMES = new Set([ + 'APIConnectionTimeoutError', + 'TimeoutError', + 'ConnectTimeoutError', + 'HeadersTimeoutError', + 'BodyTimeoutError', +]); +const TIMEOUT_ERROR_CODES = new Set([ + 'ETIMEDOUT', + 'UND_ERR_CONNECT_TIMEOUT', + 'UND_ERR_HEADERS_TIMEOUT', + 'UND_ERR_BODY_TIMEOUT', +]); +const CONNECTION_ERROR_NAMES = new Set(['APIConnectionError', 'SocketError']); +const CONNECTION_ERROR_CODES = new Set([ + 'ECONNRESET', + 'ECONNREFUSED', + 'EPIPE', + 'EAI_AGAIN', + 'UND_ERR_SOCKET', +]); +const TRANSIENT_STATUSES = new Set([408, 429]); + +interface ErrorShape { + name?: unknown; + code?: unknown; + cause?: unknown; + status?: unknown; + statusCode?: unknown; + response?: { status?: unknown }; + constructor?: { name?: string }; +} + +const asShape = (e: unknown): ErrorShape | undefined => + e && typeof e === 'object' ? (e as ErrorShape) : undefined; + +const matches = ( + e: ErrorShape, + names: Set, + codes: Set, +): boolean => + (typeof e.name === 'string' && names.has(e.name)) || + names.has(e.constructor?.name ?? '') || + (typeof e.code === 'string' && codes.has(e.code)); + +const upstreamStatus = (e: ErrorShape): number | undefined => { + const s = e.status ?? e.statusCode ?? e.response?.status; + return typeof s === 'number' ? s : undefined; +}; + +/** A request to an upstream provider ran out of time before it answered. */ +export const isUpstreamTimeoutError = (err: unknown): boolean => { + const e = asShape(err); + if (!e) return false; + if (matches(e, TIMEOUT_ERROR_NAMES, TIMEOUT_ERROR_CODES)) return true; + const cause = asShape(e.cause); + return ( + cause !== undefined && + matches(cause, TIMEOUT_ERROR_NAMES, TIMEOUT_ERROR_CODES) + ); +}; + +/** + * A failure worth retrying on the next poll: a timeout, a dropped connection, + * or a status the provider itself treats as temporary (408, 429, 5xx). Any + * other 4xx is the provider's verdict on the request and is not transient. + */ +export const isTransientUpstreamError = (err: unknown): boolean => { + if (isUpstreamTimeoutError(err)) return true; + const e = asShape(err); + if (!e) return false; + const status = upstreamStatus(e); + if (status !== undefined) { + return status >= 500 || TRANSIENT_STATUSES.has(status); + } + if (matches(e, CONNECTION_ERROR_NAMES, CONNECTION_ERROR_CODES)) return true; + const cause = asShape(e.cause); + return ( + cause !== undefined && + matches(cause, CONNECTION_ERROR_NAMES, CONNECTION_ERROR_CODES) + ); +}; + +// -- Upstream messages -- + +const MAX_UPSTREAM_MESSAGE_LENGTH = 300; + +/** Model-side content filters, as worded in provider failure payloads. */ +export const CONTENT_FILTER_PATTERN = + /\bnsfw\b|sensitive|content[\s_-]?policy|moderation|safety|\bunsafe\b|\bflagged\b|prohibited|\bE005\b/i; + +/** + * Strips markup and bounds length so an upstream HTML error page never rides + * through into a response body or an alarm signature. + */ +export const sanitizeUpstreamMessage = (raw: string): string => { + const text = raw + .replace(/<(style|script)[\s\S]*?<\/\1>/gi, ' ') + .replace(/<[^>]*>/g, ' ') + .replace(/\s+/g, ' ') + .trim(); + return text.length > MAX_UPSTREAM_MESSAGE_LENGTH + ? `${text.slice(0, MAX_UPSTREAM_MESSAGE_LENGTH - 3)}...` + : text; +}; diff --git a/src/docs/src/AI/txt2vid.md b/src/docs/src/AI/txt2vid.md index 97b11153b..67e8311f9 100644 --- a/src/docs/src/AI/txt2vid.md +++ b/src/docs/src/AI/txt2vid.md @@ -106,6 +106,20 @@ A `Promise` that resolves to an `HTMLVideoElement`. The element is preloaded, ha > **Note:** Video generation can take several minutes to complete. The returned promise resolves only when the video is ready, so keep your UI responsive (for example, by showing a spinner) while you wait. Each successful generation consumes the user’s AI credits in accordance with the model, duration, and resolution you request. +## Errors + +A rejection carries the error body as the backend sent it: `{ message, code }`. + +| Code | Meaning | +| --- | --- | +| `upstream_timeout` | The provider did not finish the clip within the ten minutes Puter waits for it. Arrives as HTTP 504. The request itself was fine; retry it, ideally with a shorter clip or a faster model. | +| `errorCode: moderation_flagged` | The provider's content filter refused the prompt or removed the generated video. Arrives as HTTP 400, with `code: bad_request` from most providers and `code: disallowed_value` from Veo. Change the prompt rather than retrying it as-is. | +| `upstream_bad_request` | The provider rejected a parameter, for example a frame rate the model does not support. Arrives as HTTP 400; the `message` carries the provider's reason. | +| `upstream_failed` | The provider accepted the request but generation failed on their side. Safe to retry. | +| `insufficient_funds` | Your balance cannot cover the estimated cost of the clip. Arrives as HTTP 402. | + +Other `upstream_*` codes mean the provider rejected the request or was unavailable; the `message` carries the provider's reason. + ## Examples Generate a sample clip (test mode)