Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions apps/gateway/src/test-utils/mock-openai-server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -410,6 +410,10 @@ interface MockWebhookDelivery {
}

const videoJobs = new Map<string, MockVideoJobState>();
const videoStatusResponses = new Map<
string,
{ status: number; body: Record<string, unknown> }
>();
const webhookDeliveries: MockWebhookDelivery[] = [];
const webhookStatuses = new Map<string, number>();

Expand Down Expand Up @@ -540,6 +544,7 @@ export function resetFailOnceCounter() {
export function resetMockVideoState() {
videoCounter = 0;
videoJobs.clear();
videoStatusResponses.clear();
webhookDeliveries.length = 0;
webhookStatuses.clear();
}
Expand Down Expand Up @@ -588,6 +593,14 @@ export function getMockVideo(videoId: string): MockVideoJobState | undefined {
return videoJobs.get(videoId);
}

export function setMockVideoStatusResponse(
videoId: string,
status: number,
body: Record<string, unknown>,
) {
videoStatusResponses.set(videoId, { status, body });
}

export function setMockWebhookStatus(name: string, status: number) {
webhookStatuses.set(name, status);
}
Expand Down Expand Up @@ -2796,6 +2809,11 @@ mockOpenAIServer.post("/mock-google-oauth/token", async (c) =>

mockOpenAIServer.get("/v1/videos/:id", async (c) => {
const id = c.req.param("id");
const statusResponse = videoStatusResponses.get(id);
if (statusResponse) {
c.status(statusResponse.status as Parameters<typeof c.status>[0]);
return c.json(statusResponse.body);
}
const job = videoJobs.get(id);

if (!job) {
Expand Down
75 changes: 75 additions & 0 deletions apps/gateway/src/videos/videos.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import { createGatewayApiTestHarness } from "@/test-utils/gateway-api-test-harne
import {
getMockVideo,
setMockVideoStatus,
setMockVideoStatusResponse,
} from "@/test-utils/mock-openai-server.js";

import { cdb, db, eq, tables } from "@llmgateway/db";
Expand Down Expand Up @@ -1756,6 +1757,80 @@ describe("videos", () => {
expect(logs[0].cost).toBe(0.85);
});

test("/v1/videos caps logged xAI polling error response contents", async () => {
await db.insert(tables.apiKey).values({
id: "token-id",
token: "real-token",
projectId: "project-id",
description: "Test API Key",
createdBy: "user-id",
});

await db.insert(tables.providerKey).values({
id: "provider-key-id",
token: "xai-test-token",
provider: "xai",
organizationId: "org-id",
baseUrl: mockServerUrl,
});

const createRes = await app.request("/v1/videos", {
method: "POST",
headers: {
"Content-Type": "application/json",
Authorization: "Bearer real-token",
},
body: JSON.stringify({
model: "xai/grok-imagine-video-1-5",
prompt: "A cat walking across a rooftop at sunset",
size: "1280x720",
seconds: 6,
image: { image_url: "data:image/png;base64,aGVsbG8=" },
}),
});

expect(createRes.status).toBe(200);
const created = await createRes.json();
const videoJob = await db.query.videoJob.findFirst({
where: { id: { eq: created.id } },
});
expect(videoJob).toBeTruthy();

const upstreamError = {
detail: `The input image could not be processed: ${"x".repeat(5000)}discarded-suffix`,
request_id: "upstream-request-id",
};
setMockVideoStatusResponse(videoJob!.upstreamId, 400, upstreamError);
await db
.update(tables.videoJob)
.set({
upstreamStatusResponse: {
llmgateway_poll_error_count: 4,
},
})
.where(eq(tables.videoJob.id, videoJob!.id));

await processPendingVideoJobs();

const persistedJob = await db.query.videoJob.findFirst({
where: { id: { eq: created.id } },
});
const log = await db.query.log.findFirst({
where: { requestId: { eq: videoJob!.requestId } },
});
const expectedError = `Upstream status request failed with status 400: ${JSON.stringify(upstreamError).slice(0, 4000)}`;
const statusResponse = persistedJob?.upstreamStatusResponse as Record<
string,
unknown
>;
expect(persistedJob?.status).toBe("failed");
expect(persistedJob?.error?.message).toContain(expectedError);
expect(statusResponse.llmgateway_last_poll_error).toBe(expectedError);
expect(log?.errorDetails?.responseText).toContain(expectedError);
expect(JSON.stringify(persistedJob)).not.toContain("discarded-suffix");
expect(JSON.stringify(log)).not.toContain("discarded-suffix");
});

test("/v1/videos supports completed google-vertex jobs", async () => {
const originalGoogleCloudProject = process.env.LLM_GOOGLE_CLOUD_PROJECT;
const originalRuntimeGoogleCloudProject = process.env.GOOGLE_CLOUD_PROJECT;
Expand Down
26 changes: 14 additions & 12 deletions apps/worker/src/services/video-jobs.ts
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,7 @@ const VIDEO_JOB_POLL_CLAIM_TTL_MS = 30_000;
const VIDEO_JOB_ERROR_BASE_DELAY_MS = 10_000;
const VIDEO_JOB_ERROR_MAX_DELAY_MS = 5 * 60 * 1000;
const VIDEO_JOB_MAX_POLL_ERROR_COUNT = 5;
const VIDEO_JOB_UPSTREAM_ERROR_TEXT_LIMIT = 4000;
const VIDEO_RESOLUTION_4K = "4k";
const VIDEO_RESOLUTION_HD = "hd";
const VIDEO_RESOLUTION_1080P = "1080p";
Expand Down Expand Up @@ -1260,22 +1261,23 @@ async function fetchJsonResponse(
): Promise<{
body: Record<string, unknown>;
response: Response;
responseText: string;
}> {
const response = await fetchWithSignals(url, init, UPSTREAM_FETCH_TIMEOUT_MS);
const text = await response.text();
const responseText = await response.text();

let body: Record<string, unknown> = {};
if (text.length > 0) {
if (responseText.length > 0) {
try {
body = JSON.parse(text) as Record<string, unknown>;
body = JSON.parse(responseText) as Record<string, unknown>;
} catch {
body = {
message: text,
message: responseText,
};
}
}

return { body, response };
return { body, response, responseText };
}

async function fetchAvalancheRecordInfo(
Expand Down Expand Up @@ -2437,19 +2439,19 @@ async function fetchGenericVideoStatus(
providerContext: ResolvedVideoProviderContext,
): Promise<Record<string, unknown>> {
const url = joinUrl(providerContext.baseUrl, `/v1/videos/${job.upstreamId}`);
const { body, response } = await fetchJsonResponse(url, {
const { body, response, responseText } = await fetchJsonResponse(url, {
method: "GET",
headers: getVideoProviderHeaders(job, providerContext),
});

if (!response.ok) {
const upstreamContents = responseText
.trim()
.slice(0, VIDEO_JOB_UPSTREAM_ERROR_TEXT_LIMIT);

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Avoid splitting Unicode surrogates when truncating errors

When a non-BMP character such as an emoji straddles the 4,000-code-unit boundary, slice() leaves a lone UTF-16 surrogate in upstreamContents. Serializing that value produces an unpaired \ud8xx escape, which PostgreSQL rejects when the polling catch stores it in the upstreamStatusResponse JSONB column; the error counter is therefore not advanced, the job remains stuck, and the rest of the claimed polling batch is skipped. Truncate on Unicode scalar boundaries before persisting the diagnostic.

Useful? React with 👍 / 👎.

throw new Error(
typeof body.error === "object" &&
body.error &&
"message" in body.error &&
typeof body.error.message === "string"
? body.error.message
: `Upstream status request failed with status ${response.status}`,
`Upstream status request failed with status ${response.status}${
upstreamContents ? `: ${upstreamContents}` : ""
}`,
);
}

Expand Down
Loading