fix(qoder): detect errors in SSE envelope to trigger failover
- Made wrapQoderSSE async to peek at first chunk before streaming - Parse first SSE line to check for error envelope (statusCodeValue !== 200) - If error detected, return error Response with proper HTTP status code - This triggers chatCore's !providerResponse.ok check and failover logic - Fixes issue where qoder 403 quota errors were wrapped as successful streams Before: qoder errors appeared as stream content with finish_reason: 'stop' After: qoder errors return proper HTTP error codes, triggering provider failover
This commit is contained in:
@@ -222,22 +222,75 @@ async function buildQoderRequestBody({ model, body, credentials, log, proxyOptio
|
|||||||
* and re-emit as `data: <inner>\n\n`. Errors become `data: [DONE]\n\n` plus
|
* and re-emit as `data: <inner>\n\n`. Errors become `data: [DONE]\n\n` plus
|
||||||
* a synthetic OpenAI error chunk.
|
* a synthetic OpenAI error chunk.
|
||||||
*/
|
*/
|
||||||
function wrapQoderSSE(response, model) {
|
async function wrapQoderSSE(response, model) {
|
||||||
if (!response.ok || !response.body) return response;
|
if (!response.ok || !response.body) return response;
|
||||||
|
|
||||||
const decoder = new TextDecoder();
|
const decoder = new TextDecoder();
|
||||||
const encoder = new TextEncoder();
|
const encoder = new TextEncoder();
|
||||||
|
|
||||||
|
// Peek at first chunk to detect errors early
|
||||||
|
const reader = response.body.getReader();
|
||||||
|
const firstRead = await reader.read();
|
||||||
|
|
||||||
|
if (firstRead.done) {
|
||||||
|
// Empty stream
|
||||||
|
return new Response("data: [DONE]\n\n", {
|
||||||
|
status: response.status,
|
||||||
|
statusText: response.statusText,
|
||||||
|
headers: {
|
||||||
|
"Content-Type": "text/event-stream",
|
||||||
|
"Cache-Control": "no-cache",
|
||||||
|
},
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
// Parse first line to check for error
|
||||||
|
const firstText = decoder.decode(firstRead.value, { stream: true });
|
||||||
|
const nlIndex = firstText.indexOf("\n");
|
||||||
|
const firstLine = nlIndex !== -1 ? firstText.slice(0, nlIndex) : firstText;
|
||||||
|
const trimmed = firstLine.replace(/\r$/, "").trim();
|
||||||
|
|
||||||
|
if (trimmed.startsWith("data:")) {
|
||||||
|
const data = trimmed.slice(5).trimStart();
|
||||||
|
if (data !== "[DONE]") {
|
||||||
|
try {
|
||||||
|
const envelope = JSON.parse(data);
|
||||||
|
const statusVal = typeof envelope.statusCodeValue === "number" ? envelope.statusCodeValue : 200;
|
||||||
|
|
||||||
|
if (statusVal !== 200) {
|
||||||
|
// Error detected - return error Response to trigger failover
|
||||||
|
const msg = envelope.body || `upstream status ${statusVal}`;
|
||||||
|
const errorResponse = new Response(
|
||||||
|
JSON.stringify({
|
||||||
|
error: {
|
||||||
|
message: `qoder error ${statusVal}: ${truncate(msg, 500)}`,
|
||||||
|
type: "upstream_error",
|
||||||
|
code: String(statusVal)
|
||||||
|
}
|
||||||
|
}),
|
||||||
|
{
|
||||||
|
status: statusVal >= 400 && statusVal < 600 ? statusVal : 502,
|
||||||
|
headers: { "Content-Type": "application/json" }
|
||||||
|
}
|
||||||
|
);
|
||||||
|
reader.cancel();
|
||||||
|
return errorResponse;
|
||||||
|
}
|
||||||
|
} catch (e) {
|
||||||
|
// Not JSON, continue as normal stream
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// No error detected - proceed with normal TransformStream
|
||||||
let buffer = "";
|
let buffer = "";
|
||||||
let doneEmitted = false;
|
let doneEmitted = false;
|
||||||
|
|
||||||
// Process one already-extracted SSE line (no trailing newline). Returns
|
|
||||||
// false when the line indicated end-of-stream so the caller can stop
|
|
||||||
// forwarding any remaining chunks after [DONE].
|
|
||||||
const processLine = (line, controller) => {
|
const processLine = (line, controller) => {
|
||||||
const trimmed = line.replace(/\r$/, "").trim();
|
const trimmed = line.replace(/\r$/, "").trim();
|
||||||
if (!trimmed) return;
|
if (!trimmed) return;
|
||||||
if (!trimmed.startsWith("data:")) return;
|
if (!trimmed.startsWith("data:")) return;
|
||||||
if (doneEmitted) return; // never forward chunks past stream end
|
if (doneEmitted) return;
|
||||||
|
|
||||||
const data = trimmed.slice(5).trimStart();
|
const data = trimmed.slice(5).trimStart();
|
||||||
if (data === "[DONE]") {
|
if (data === "[DONE]") {
|
||||||
@@ -270,10 +323,6 @@ function wrapQoderSSE(response, model) {
|
|||||||
doneEmitted = true;
|
doneEmitted = true;
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
// Inner is an OpenAI-shaped chunk. Strip any embedded newlines so the
|
|
||||||
// SSE frame stays a single event (a literal "\n" inside `inner` would
|
|
||||||
// otherwise split the frame across multiple data: lines and downstream
|
|
||||||
// parsers would reassemble them as separate events).
|
|
||||||
const sanitized = inner.replace(/\r?\n/g, "");
|
const sanitized = inner.replace(/\r?\n/g, "");
|
||||||
controller.enqueue(encoder.encode(`data: ${sanitized}\n\n`));
|
controller.enqueue(encoder.encode(`data: ${sanitized}\n\n`));
|
||||||
};
|
};
|
||||||
@@ -289,13 +338,7 @@ function wrapQoderSSE(response, model) {
|
|||||||
}
|
}
|
||||||
},
|
},
|
||||||
flush(controller) {
|
flush(controller) {
|
||||||
// Finalize the decoder so any pending multi-byte sequence is
|
|
||||||
// released into `buffer` instead of being silently dropped.
|
|
||||||
buffer += decoder.decode();
|
buffer += decoder.decode();
|
||||||
// Drain any trailing line that arrived without a terminating newline
|
|
||||||
// (e.g. upstream closed the socket immediately after the last write,
|
|
||||||
// or a CDN stripped the final CRLF). Without this, the chunk that
|
|
||||||
// carries finish_reason is silently lost.
|
|
||||||
if (buffer.length > 0) {
|
if (buffer.length > 0) {
|
||||||
processLine(buffer, controller);
|
processLine(buffer, controller);
|
||||||
buffer = "";
|
buffer = "";
|
||||||
@@ -307,9 +350,45 @@ function wrapQoderSSE(response, model) {
|
|||||||
},
|
},
|
||||||
});
|
});
|
||||||
|
|
||||||
const transformed = response.body.pipeThrough(transform);
|
// Create a ReadableStream that emits the first chunk + remaining chunks
|
||||||
// Build a Response with passable headers; the streaming handler reads
|
const combinedStream = new ReadableStream({
|
||||||
// `.body` as a ReadableStream regardless of Content-Type.
|
start(controller) {
|
||||||
|
controller.enqueue(firstRead.value);
|
||||||
|
},
|
||||||
|
async pull(controller) {
|
||||||
|
const { done, value } = await reader.read();
|
||||||
|
if (done) {
|
||||||
|
controller.close();
|
||||||
|
} else {
|
||||||
|
controller.enqueue(value);
|
||||||
|
}
|
||||||
|
},
|
||||||
|
cancel() {
|
||||||
|
reader.cancel();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
const transformed = combinedStream.pipeThrough(transform);
|
||||||
|
return new Response(transformed, {
|
||||||
|
status: response.status,
|
||||||
|
statusText: response.statusText,
|
||||||
|
headers: {
|
||||||
|
"Content-Type": "text/event-stream",
|
||||||
|
"Cache-Control": "no-cache",
|
||||||
|
},
|
||||||
|
});
|
||||||
|
}
|
||||||
|
controller.close();
|
||||||
|
} else {
|
||||||
|
controller.enqueue(value);
|
||||||
|
}
|
||||||
|
},
|
||||||
|
cancel() {
|
||||||
|
reader.cancel();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
const transformed = remainingStream.pipeThrough(transform);
|
||||||
return new Response(transformed, {
|
return new Response(transformed, {
|
||||||
status: response.status,
|
status: response.status,
|
||||||
statusText: response.statusText,
|
statusText: response.statusText,
|
||||||
@@ -430,7 +509,7 @@ export class QoderExecutor extends BaseExecutor {
|
|||||||
return { response, url, headers, transformedBody: payload };
|
return { response, url, headers, transformedBody: payload };
|
||||||
}
|
}
|
||||||
|
|
||||||
const wrapped = wrapQoderSSE(response, `qoder/${qoderKey}`);
|
const wrapped = await wrapQoderSSE(response, `qoder/${qoderKey}`);
|
||||||
return { response: wrapped, url, headers, transformedBody: payload };
|
return { response: wrapped, url, headers, transformedBody: payload };
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user