Skip to content

Commit dd3def7

Browse files
Alix-007Alix-007
committed
fix(voyage): bound embedding-batch status, error, and non-OK responses
The batch status read (fetchVoyageBatchStatus) parsed its response with an unbounded await res.json(), and the batch error-file read (readVoyageBatchError) buffered the whole body via await res.text(). On top of that, the non-OK (4xx/5xx) diagnostic body was still read unbounded: assertVoyageResponseOk did await res.text() before throwing, and the non-OK output-file branch in runVoyageEmbeddingBatches did the same. Voyage base URLs are user-supplied and reachable via SSRF, so a misbehaving or hostile endpoint could stream an unbounded body into memory on any of these paths before parsing. Route the status JSON through the shared readProviderJsonResponse, the error file through readResponseWithLimit, and now the non-OK diagnostic body through readResponseWithLimit as well, all under a single 16 MiB cap, cancelling the stream on overflow before decode/parse. assertVoyageResponseOk preserves its original "${context}: ${status} ${text}" diagnostic shape for under-cap bodies and throws a bounded "(error body exceeds <N> bytes)" on overflow; the non-OK output-file branch now reuses it instead of a duplicate unbounded read. The existing error-file fail-soft handling (formatUnavailableBatchError) is preserved, so a capped endpoint degrades gracefully. The submit path already bounds its body via postJsonWithRetry/maxResponseBytes and is left untouched. Symmetric counterpart to the #96027/#96038 response-limit campaign.
1 parent 3ab8d6a commit dd3def7

2 files changed

Lines changed: 262 additions & 10 deletions

File tree

‎extensions/voyage/embedding-batch.test.ts‎

Lines changed: 217 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,217 @@
1+
// Voyage batch tests cover bounded status/error response reads.
2+
import { describe, expect, it } from "vitest";
3+
import type { VoyageEmbeddingClient } from "./embedding-provider.js";
4+
import { testing } from "./embedding-batch.js";
5+
6+
const { fetchVoyageBatchStatus, readVoyageBatchError, VOYAGE_BATCH_RESPONSE_MAX_BYTES } = testing;
7+
8+
function buildClient(): VoyageEmbeddingClient {
9+
return {
10+
baseUrl: "https://api.voyageai.test/v1",
11+
headers: { authorization: "Bearer test" },
12+
model: "voyage-3",
13+
};
14+
}
15+
16+
/**
17+
* Build deps whose withRemoteHttpResponse drives the real onResponse against a
18+
* caller-provided Response, so the bounded readers run exactly as in production.
19+
*/
20+
function buildDeps(response: Response): Parameters<typeof fetchVoyageBatchStatus>[0]["deps"] {
21+
return {
22+
now: () => 0,
23+
sleep: async () => {},
24+
postJsonWithRetry: (async () => {
25+
throw new Error("postJsonWithRetry should not be called in these tests");
26+
}) as never,
27+
uploadBatchJsonlFile: (async () => {
28+
throw new Error("uploadBatchJsonlFile should not be called in these tests");
29+
}) as never,
30+
withRemoteHttpResponse: (async (params: { onResponse: (res: Response) => Promise<unknown> }) =>
31+
await params.onResponse(response)) as never,
32+
};
33+
}
34+
35+
/**
36+
* A streaming JSON-ish body that proves an oversized response stops being read
37+
* before the whole advertised payload is buffered into memory. getReadCount
38+
* reports how many chunks were pulled; cancel() flips wasCanceled.
39+
*/
40+
function streamingResponse(params: { chunkCount: number; chunkSize: number; status?: number }): {
41+
response: Response;
42+
getReadCount: () => number;
43+
wasCanceled: () => boolean;
44+
} {
45+
let reads = 0;
46+
let canceled = false;
47+
const encoder = new TextEncoder();
48+
const stream = new ReadableStream<Uint8Array>({
49+
pull(controller) {
50+
if (reads >= params.chunkCount) {
51+
controller.close();
52+
return;
53+
}
54+
reads += 1;
55+
controller.enqueue(encoder.encode("a".repeat(params.chunkSize)));
56+
},
57+
cancel() {
58+
canceled = true;
59+
},
60+
});
61+
return {
62+
response: new Response(stream, {
63+
status: params.status ?? 200,
64+
headers: { "content-type": "application/json" },
65+
}),
66+
getReadCount: () => reads,
67+
wasCanceled: () => canceled,
68+
};
69+
}
70+
71+
describe("voyage batch bounded reads", () => {
72+
it("uses a 16 MiB cap for batch status/error responses", () => {
73+
expect(VOYAGE_BATCH_RESPONSE_MAX_BYTES).toBe(16 * 1024 * 1024);
74+
});
75+
76+
it("parses a well-formed batch status response under the byte cap", async () => {
77+
const response = new Response(JSON.stringify({ id: "batch_1", status: "completed" }), {
78+
status: 200,
79+
headers: { "content-type": "application/json" },
80+
});
81+
82+
const status = await fetchVoyageBatchStatus({
83+
client: buildClient(),
84+
batchId: "batch_1",
85+
deps: buildDeps(response),
86+
});
87+
88+
expect(status).toEqual({ id: "batch_1", status: "completed" });
89+
});
90+
91+
it("caps an oversized batch status stream instead of buffering the whole body", async () => {
92+
const streamed = streamingResponse({ chunkCount: 64, chunkSize: 1024 });
93+
94+
await expect(
95+
fetchVoyageBatchStatus({
96+
client: buildClient(),
97+
batchId: "batch_1",
98+
deps: buildDeps(streamed.response),
99+
maxResponseBytes: 4096,
100+
}),
101+
).rejects.toThrow(/voyage-batch-status: JSON response exceeds 4096 bytes/);
102+
103+
// Stream was cancelled mid-flight: fewer chunks read than the full payload.
104+
expect(streamed.getReadCount()).toBeLessThan(64);
105+
expect(streamed.wasCanceled()).toBe(true);
106+
});
107+
108+
it("preserves the full NDJSON parse chain for an under-cap error file", async () => {
109+
// Multi-line NDJSON with a blank line proves the bounded read does not
110+
// disturb the original trim/split("\n")/JSON.parse/extractBatchErrorMessage
111+
// pipeline: the first useful error message is still extracted byte-for-byte
112+
// identically to the pre-change `await res.text()` path.
113+
const body = [
114+
JSON.stringify({ custom_id: "req-0", response: { status_code: 200 } }),
115+
"",
116+
JSON.stringify({ custom_id: "req-1", error: { message: "voyage upstream rejected" } }),
117+
JSON.stringify({ custom_id: "req-2", error: { message: "second error ignored" } }),
118+
"",
119+
].join("\n");
120+
const response = new Response(body, {
121+
status: 200,
122+
headers: { "content-type": "application/x-ndjson" },
123+
});
124+
125+
const message = await readVoyageBatchError({
126+
client: buildClient(),
127+
errorFileId: "file_1",
128+
deps: buildDeps(response),
129+
});
130+
131+
// extractBatchErrorMessage returns the first line carrying a message, so the
132+
// success line is skipped and the second error is not surfaced.
133+
expect(message).toBe("voyage upstream rejected");
134+
});
135+
136+
it("returns undefined for an empty error file via the original empty-body branch", async () => {
137+
// Whitespace-only body must still hit the `!text.trim()` short-circuit after
138+
// decoding the bounded buffer, returning undefined exactly as before.
139+
const response = new Response(" \n", {
140+
status: 200,
141+
headers: { "content-type": "application/x-ndjson" },
142+
});
143+
144+
const message = await readVoyageBatchError({
145+
client: buildClient(),
146+
errorFileId: "file_1",
147+
deps: buildDeps(response),
148+
});
149+
150+
expect(message).toBeUndefined();
151+
});
152+
153+
it("fail-softs an oversized error file into formatUnavailableBatchError by design", async () => {
154+
const streamed = streamingResponse({ chunkCount: 64, chunkSize: 1024 });
155+
156+
// Intended behavior: an over-cap error file must NOT throw out of
157+
// readVoyageBatchError. An unbounded error body would otherwise OOM the
158+
// worker, so the bounded overflow error is caught and degraded into a
159+
// diagnostic string via formatUnavailableBatchError. We accept the lost
160+
// detail; the overflow message names the cap so the truncation is visible.
161+
const readError = async () =>
162+
await readVoyageBatchError({
163+
client: buildClient(),
164+
errorFileId: "file_1",
165+
deps: buildDeps(streamed.response),
166+
maxResponseBytes: 4096,
167+
});
168+
169+
await expect(readError()).resolves.toMatch(
170+
/error file unavailable: voyage batch error file content exceeds 4096 bytes/,
171+
);
172+
173+
// The bounded reader still cancels the stream mid-flight rather than
174+
// buffering the whole advertised payload before failing soft.
175+
expect(streamed.getReadCount()).toBeLessThan(64);
176+
expect(streamed.wasCanceled()).toBe(true);
177+
});
178+
179+
it("caps an oversized non-OK (error) diagnostic body instead of buffering it whole", async () => {
180+
// Regression for the non-OK gap: `assertVoyageResponseOk` previously read the
181+
// 4xx/5xx diagnostic body with an unbounded `await res.text()`. A hostile
182+
// endpoint can return a 500 with a never-ending body, so that read must be
183+
// bounded too. Drive a streaming 500 through the real status path and assert
184+
// the bounded overflow error fires and the stream is cancelled mid-flight.
185+
const streamed = streamingResponse({ chunkCount: 64, chunkSize: 1024, status: 500 });
186+
187+
await expect(
188+
fetchVoyageBatchStatus({
189+
client: buildClient(),
190+
batchId: "batch_1",
191+
deps: buildDeps(streamed.response),
192+
maxResponseBytes: 4096,
193+
}),
194+
).rejects.toThrow(/voyage batch status failed: 500 \(error body exceeds 4096 bytes\)/);
195+
196+
// Stream was cancelled mid-flight rather than draining the whole body.
197+
expect(streamed.getReadCount()).toBeLessThan(64);
198+
expect(streamed.wasCanceled()).toBe(true);
199+
});
200+
201+
it("preserves the diagnostic shape for a small non-OK (error) body", async () => {
202+
// Under-cap non-OK body must still surface the original
203+
// `${context}: ${status} ${text}` diagnostic byte-for-byte.
204+
const response = new Response("voyage upstream is down", {
205+
status: 503,
206+
headers: { "content-type": "text/plain" },
207+
});
208+
209+
await expect(
210+
fetchVoyageBatchStatus({
211+
client: buildClient(),
212+
batchId: "batch_1",
213+
deps: buildDeps(response),
214+
}),
215+
).rejects.toThrow(/voyage batch status failed: 503 voyage upstream is down/);
216+
});
217+
});

‎extensions/voyage/embedding-batch.ts‎

Lines changed: 45 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@ import {
2121
uploadBatchJsonlFile,
2222
withRemoteHttpResponse,
2323
} from "openclaw/plugin-sdk/memory-core-host-engine-embeddings";
24+
import { readProviderJsonResponse } from "openclaw/plugin-sdk/provider-http";
25+
import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime";
2426
import { normalizeStringEntries } from "openclaw/plugin-sdk/string-coerce-runtime";
2527
import type { VoyageEmbeddingClient } from "./embedding-provider.js";
2628

@@ -41,6 +43,10 @@ type VoyageBatchOutputLine = ProviderBatchOutputLine;
4143
const VOYAGE_BATCH_ENDPOINT = EMBEDDING_BATCH_ENDPOINT;
4244
const VOYAGE_BATCH_COMPLETION_WINDOW = "12h";
4345
const VOYAGE_BATCH_MAX_REQUESTS = 50000;
46+
// Voyage batch status/error responses are untrusted external bodies. Cap them
47+
// the same way other bundled providers do (16 MiB) so a misbehaving or hostile
48+
// endpoint cannot stream an unbounded body into memory before we parse it.
49+
const VOYAGE_BATCH_RESPONSE_MAX_BYTES = 16 * 1024 * 1024;
4450

4551
type VoyageBatchDeps = {
4652
now: () => number;
@@ -65,9 +71,23 @@ function resolveVoyageBatchDeps(overrides: Partial<VoyageBatchDeps> | undefined)
6571
};
6672
}
6773

68-
async function assertVoyageResponseOk(res: Response, context: string): Promise<void> {
74+
async function assertVoyageResponseOk(
75+
res: Response,
76+
context: string,
77+
maxBytes: number = VOYAGE_BATCH_RESPONSE_MAX_BYTES,
78+
): Promise<void> {
6979
if (!res.ok) {
70-
const text = await res.text();
80+
// The non-OK diagnostic body is just as untrusted as the success body: a
81+
// misbehaving or hostile endpoint can return a 4xx/5xx with an unbounded
82+
// body, and the old `await res.text()` buffered it whole before we threw.
83+
// Read it through the same bounded reader (16 MiB cap, stream cancelled on
84+
// overflow) while preserving the original `${context}: ${status} ${text}`
85+
// diagnostic shape for backward compatibility.
86+
const bytes = await readResponseWithLimit(res, maxBytes, {
87+
onOverflow: ({ maxBytes: maxBytesLocal }) =>
88+
new Error(`${context}: ${res.status} (error body exceeds ${maxBytesLocal} bytes)`),
89+
});
90+
const text = new TextDecoder().decode(bytes);
7191
throw new Error(`${context}: ${res.status} ${text}`);
7292
}
7393
}
@@ -127,14 +147,18 @@ async function fetchVoyageBatchStatus(params: {
127147
client: VoyageEmbeddingClient;
128148
batchId: string;
129149
deps: VoyageBatchDeps;
150+
maxResponseBytes?: number;
130151
}): Promise<VoyageBatchStatus> {
152+
const maxBytes = params.maxResponseBytes ?? VOYAGE_BATCH_RESPONSE_MAX_BYTES;
131153
return await params.deps.withRemoteHttpResponse(
132154
buildVoyageBatchRequest({
133155
client: params.client,
134156
path: `batches/${params.batchId}`,
135157
onResponse: async (res) => {
136-
await assertVoyageResponseOk(res, "voyage batch status failed");
137-
return (await res.json()) as VoyageBatchStatus;
158+
await assertVoyageResponseOk(res, "voyage batch status failed", maxBytes);
159+
return await readProviderJsonResponse<VoyageBatchStatus>(res, "voyage-batch-status", {
160+
maxBytes,
161+
});
138162
},
139163
}),
140164
);
@@ -144,15 +168,21 @@ async function readVoyageBatchError(params: {
144168
client: VoyageEmbeddingClient;
145169
errorFileId: string;
146170
deps: VoyageBatchDeps;
171+
maxResponseBytes?: number;
147172
}): Promise<string | undefined> {
173+
const maxBytes = params.maxResponseBytes ?? VOYAGE_BATCH_RESPONSE_MAX_BYTES;
148174
try {
149175
return await params.deps.withRemoteHttpResponse(
150176
buildVoyageBatchRequest({
151177
client: params.client,
152178
path: `files/${params.errorFileId}/content`,
153179
onResponse: async (res) => {
154-
await assertVoyageResponseOk(res, "voyage batch error file content failed");
155-
const text = await res.text();
180+
await assertVoyageResponseOk(res, "voyage batch error file content failed", maxBytes);
181+
const bytes = await readResponseWithLimit(res, maxBytes, {
182+
onOverflow: ({ maxBytes: maxBytesLocal }) =>
183+
new Error(`voyage batch error file content exceeds ${maxBytesLocal} bytes`),
184+
});
185+
const text = new TextDecoder().decode(bytes);
156186
if (!text.trim()) {
157187
return undefined;
158188
}
@@ -280,10 +310,9 @@ export async function runVoyageEmbeddingBatches(
280310
headers: buildBatchHeaders(params.client, { json: true }),
281311
},
282312
onResponse: async (contentRes) => {
283-
if (!contentRes.ok) {
284-
const text = await contentRes.text();
285-
throw new Error(`voyage batch file content failed: ${contentRes.status} ${text}`);
286-
}
313+
// Same bounded non-OK diagnostic read as the status/error-file paths:
314+
// the failure body is untrusted, so cap it instead of `await text()`.
315+
await assertVoyageResponseOk(contentRes, "voyage batch file content failed");
287316

288317
if (!contentRes.body) {
289318
return;
@@ -316,3 +345,9 @@ export async function runVoyageEmbeddingBatches(
316345
},
317346
});
318347
}
348+
349+
export const testing = {
350+
fetchVoyageBatchStatus,
351+
readVoyageBatchError,
352+
VOYAGE_BATCH_RESPONSE_MAX_BYTES,
353+
} as const;

0 commit comments

Comments
 (0)