mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-08-15 03:32:21 +03:00
fix(sse): stop the executor-contract guard from hot-looping the router (#10373)
The `instanceof Response` guard from #10256 broke two ways: 1. `instanceof` is nominal against `globalThis.Response`, but proxyFetch dispatches through the npm undici package's fetch, whose Response is a different class — so valid upstream responses were rejected as contract violations. Replaced with `isResponseLike()` (instanceof fast path + structural brand/member probe); genuinely malformed shapes still throw. 2. The thrown error had no `.status`, so it fell through to chatCore's BAD_GATEWAY default — an internal defect was treated as a flaky provider, cooling the connection down and retrying forever. It now carries status 500 + `executor_contract_violation`, registered as request-scoped and terminal (no cooldown, no breaker, no retry). batch_api.test.ts went from exit 124 (infinite hang, pinning Unit shard 4/4 in every open PR) to exit 0, 22/22 passing. Closes #10360
This commit is contained in:
committed by
GitHub
parent
27e163e2c9
commit
90458a613c
@@ -179,6 +179,24 @@ export const HTTP_STATUS = {
|
|||||||
SERVICE_UNAVAILABLE: 503,
|
SERVICE_UNAVAILABLE: 503,
|
||||||
GATEWAY_TIMEOUT: 504,
|
GATEWAY_TIMEOUT: 504,
|
||||||
};
|
};
|
||||||
|
|
||||||
|
/**
|
||||||
|
* #10360 — stable error code for an INTERNAL violation of the executor
|
||||||
|
* `execute()` result contract (`normalizeExecutorResult` received something
|
||||||
|
* that is neither a Response nor `{ response: Response }`).
|
||||||
|
*
|
||||||
|
* This is our own bug, never a provider/account health signal, so every
|
||||||
|
* resilience layer must treat it as request-scoped and terminal: no connection
|
||||||
|
* cooldown, no provider circuit-breaker trip, no retry. It rides on the error's
|
||||||
|
* `.code` (read by `getUpstreamErrorIdentifier`) and therefore reaches
|
||||||
|
* `checkFallbackError` as `structuredError.code` and the chat/combo predicates
|
||||||
|
* as `result.errorCode`.
|
||||||
|
*
|
||||||
|
* Lives here (leaf config module) so both `open-sse/handlers/` and
|
||||||
|
* `open-sse/services/` can import it without creating a cycle.
|
||||||
|
*/
|
||||||
|
export const EXECUTOR_CONTRACT_VIOLATION_CODE = "executor_contract_violation";
|
||||||
|
|
||||||
export {
|
export {
|
||||||
BACKOFF_CONFIG,
|
BACKOFF_CONFIG,
|
||||||
COOLDOWN_MS,
|
COOLDOWN_MS,
|
||||||
|
|||||||
@@ -1,4 +1,8 @@
|
|||||||
import { FETCH_TIMEOUT_MS } from "../../config/constants.ts";
|
import {
|
||||||
|
EXECUTOR_CONTRACT_VIOLATION_CODE,
|
||||||
|
FETCH_TIMEOUT_MS,
|
||||||
|
HTTP_STATUS,
|
||||||
|
} from "../../config/constants.ts";
|
||||||
import { getModelTimeoutMs } from "../../config/providerModels.ts";
|
import { getModelTimeoutMs } from "../../config/providerModels.ts";
|
||||||
import {
|
import {
|
||||||
getLoggedInputTokens,
|
getLoggedInputTokens,
|
||||||
@@ -98,6 +102,62 @@ export function getExecutorTimeoutMs(executor: unknown, provider?: string, model
|
|||||||
return resolveProviderTimeoutMs(executor);
|
return resolveProviderTimeoutMs(executor);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Cross-realm Response detection (#10360).
|
||||||
|
*
|
||||||
|
* `instanceof Response` is a NOMINAL check against `globalThis.Response`, and
|
||||||
|
* OmniRoute's default egress does not use the global one: `proxyFetch.ts`
|
||||||
|
* dispatches through the npm `undici` package's `fetch`, whose `Response` is a
|
||||||
|
* different class from the Node built-in. A bare `instanceof` therefore
|
||||||
|
* rejected virtually every real upstream response as a "contract violation".
|
||||||
|
*
|
||||||
|
* Accept the built-in fast path first, then fall back to a structural probe:
|
||||||
|
* the `Symbol.toStringTag` brand plus the members the pipeline actually reads
|
||||||
|
* (`status`/`ok`/`headers.get`/`text`/`clone`). A plain `{ status, ok }` bag
|
||||||
|
* still fails, so the guard keeps its value.
|
||||||
|
*/
|
||||||
|
export function isResponseLike(value: unknown): value is Response {
|
||||||
|
if (value instanceof Response) return true;
|
||||||
|
if (!value || typeof value !== "object") return false;
|
||||||
|
const candidate = value as {
|
||||||
|
status?: unknown;
|
||||||
|
ok?: unknown;
|
||||||
|
headers?: { get?: unknown } | null;
|
||||||
|
text?: unknown;
|
||||||
|
clone?: unknown;
|
||||||
|
};
|
||||||
|
return (
|
||||||
|
Object.prototype.toString.call(value) === "[object Response]" &&
|
||||||
|
typeof candidate.status === "number" &&
|
||||||
|
typeof candidate.ok === "boolean" &&
|
||||||
|
!!candidate.headers &&
|
||||||
|
typeof candidate.headers.get === "function" &&
|
||||||
|
typeof candidate.text === "function" &&
|
||||||
|
typeof candidate.clone === "function"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Builds the terminal error thrown on a genuine contract violation (#10360).
|
||||||
|
*
|
||||||
|
* Carries `status = 500` and `code = EXECUTOR_CONTRACT_VIOLATION_CODE` so the
|
||||||
|
* failure is classified as an INTERNAL, non-retryable defect instead of falling
|
||||||
|
* through chatCore's `BAD_GATEWAY` default. A 502 made every layer treat our own
|
||||||
|
* bug as a flaky provider: the connection was cooled down as "rate limited", the
|
||||||
|
* provider breaker counted it, and the batch runner (which retries 429/502/504)
|
||||||
|
* span for its full 24h window on an error that can never resolve itself.
|
||||||
|
*/
|
||||||
|
export function createExecutorContractError(): Error & { status: number; code: string } {
|
||||||
|
const err = new TypeError("Executor result must contain a Response") as TypeError & {
|
||||||
|
status: number;
|
||||||
|
code: string;
|
||||||
|
};
|
||||||
|
err.name = "ExecutorContractError";
|
||||||
|
err.status = HTTP_STATUS.SERVER_ERROR;
|
||||||
|
err.code = EXECUTOR_CONTRACT_VIOLATION_CODE;
|
||||||
|
return err;
|
||||||
|
}
|
||||||
|
|
||||||
export function normalizeExecutorResult(result: unknown): {
|
export function normalizeExecutorResult(result: unknown): {
|
||||||
response: Response;
|
response: Response;
|
||||||
url: string;
|
url: string;
|
||||||
@@ -105,16 +165,16 @@ export function normalizeExecutorResult(result: unknown): {
|
|||||||
transformedBody: unknown;
|
transformedBody: unknown;
|
||||||
transport?: string;
|
transport?: string;
|
||||||
} {
|
} {
|
||||||
if (result instanceof Response) {
|
if (isResponseLike(result)) {
|
||||||
return { response: result, url: "", headers: {}, transformedBody: null };
|
return { response: result, url: "", headers: {}, transformedBody: null };
|
||||||
}
|
}
|
||||||
if (
|
if (
|
||||||
!result ||
|
!result ||
|
||||||
typeof result !== "object" ||
|
typeof result !== "object" ||
|
||||||
!("response" in result) ||
|
!("response" in result) ||
|
||||||
!(result.response instanceof Response)
|
!isResponseLike(result.response)
|
||||||
) {
|
) {
|
||||||
throw new TypeError("Executor result must contain a Response");
|
throw createExecutorContractError();
|
||||||
}
|
}
|
||||||
const normalized = result as {
|
const normalized = result as {
|
||||||
response: Response;
|
response: Response;
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
import {
|
import {
|
||||||
BACKOFF_STEPS_MS,
|
BACKOFF_STEPS_MS,
|
||||||
|
EXECUTOR_CONTRACT_VIOLATION_CODE,
|
||||||
PROVIDER_PROFILES,
|
PROVIDER_PROFILES,
|
||||||
RateLimitReason,
|
RateLimitReason,
|
||||||
HTTP_STATUS,
|
HTTP_STATUS,
|
||||||
@@ -1458,6 +1459,21 @@ export function checkFallbackError(
|
|||||||
* caller can persist an explicit reset window instead of the engine's scaled cooldown. */
|
* caller can persist an explicit reset window instead of the engine's scaled cooldown. */
|
||||||
configuredCooldownMs?: number;
|
configuredCooldownMs?: number;
|
||||||
} {
|
} {
|
||||||
|
// #10360: an executor-result contract violation is OUR bug, not the provider's.
|
||||||
|
// Retrying reproduces it verbatim, and cooling the connection down (or tripping
|
||||||
|
// the provider breaker) punishes a healthy account for an internal defect. Must
|
||||||
|
// run before every other classification — the surfaced status is a plain 500,
|
||||||
|
// which the retryable set below would otherwise treat as a transient upstream
|
||||||
|
// failure and hand a backoff cooldown.
|
||||||
|
if (structuredError?.code === EXECUTOR_CONTRACT_VIOLATION_CODE) {
|
||||||
|
return {
|
||||||
|
shouldFallback: false,
|
||||||
|
cooldownMs: 0,
|
||||||
|
reason: EXECUTOR_CONTRACT_VIOLATION_CODE,
|
||||||
|
skipProviderBreaker: true,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
const svc = serviceSupervisorCooldown(status, headers);
|
const svc = serviceSupervisorCooldown(status, headers);
|
||||||
if (svc) return svc;
|
if (svc) return svc;
|
||||||
const rg = rot.gateFor(status, rotation?.account);
|
const rg = rot.gateFor(status, rotation?.account);
|
||||||
|
|||||||
@@ -6,6 +6,7 @@
|
|||||||
* predicates are re-exported from combo.ts for backward compatibility.
|
* predicates are re-exported from combo.ts for backward compatibility.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
|
import { EXECUTOR_CONTRACT_VIOLATION_CODE } from "../../config/constants.ts";
|
||||||
import { errorResponse } from "../../utils/error.ts";
|
import { errorResponse } from "../../utils/error.ts";
|
||||||
import { parseModel } from "../model.ts";
|
import { parseModel } from "../model.ts";
|
||||||
import { isSelfInflictedUpstreamTimeout } from "../../handlers/chatCore/cooldownClassification.ts";
|
import { isSelfInflictedUpstreamTimeout } from "../../handlers/chatCore/cooldownClassification.ts";
|
||||||
@@ -201,6 +202,9 @@ const REQUEST_SCOPED_UPSTREAM_ERROR_CODES: Record<string, true> = {
|
|||||||
rate_limit_queue_timeout: true,
|
rate_limit_queue_timeout: true,
|
||||||
rate_limit_queue_full: true,
|
rate_limit_queue_full: true,
|
||||||
rate_limit_queue_wedged: true,
|
rate_limit_queue_wedged: true,
|
||||||
|
// #10360: our own executor-result contract violation. An internal defect, not
|
||||||
|
// a provider/account fault — it must never cool a connection or trip a breaker.
|
||||||
|
[EXECUTOR_CONTRACT_VIOLATION_CODE]: true,
|
||||||
};
|
};
|
||||||
|
|
||||||
/** Request/model-specific failures must not poison provider-wide resilience state. */
|
/** Request/model-specific failures must not poison provider-wide resilience state. */
|
||||||
|
|||||||
@@ -219,6 +219,7 @@
|
|||||||
"tests/unit/edgetts-provider.test.ts",
|
"tests/unit/edgetts-provider.test.ts",
|
||||||
"tests/unit/embeddings-auth.test.ts",
|
"tests/unit/embeddings-auth.test.ts",
|
||||||
"tests/unit/error-classification.test.ts",
|
"tests/unit/error-classification.test.ts",
|
||||||
|
"tests/unit/executor-contract-violation-terminal.test.ts",
|
||||||
"tests/unit/error-message-sanitization.test.ts",
|
"tests/unit/error-message-sanitization.test.ts",
|
||||||
"tests/unit/error-sensitive-redaction.test.ts",
|
"tests/unit/error-sensitive-redaction.test.ts",
|
||||||
"tests/unit/execute-chat-resource-pressure-breaker.test.ts",
|
"tests/unit/execute-chat-resource-pressure-breaker.test.ts",
|
||||||
|
|||||||
173
tests/unit/executor-contract-violation-terminal.test.ts
Normal file
173
tests/unit/executor-contract-violation-terminal.test.ts
Normal file
@@ -0,0 +1,173 @@
|
|||||||
|
/**
|
||||||
|
* #10360 — the executor-result contract guard must not hot-loop the router.
|
||||||
|
*
|
||||||
|
* Two defects, one symptom (`tests/unit/batch_api.test.ts` hanging forever):
|
||||||
|
*
|
||||||
|
* 1. CROSS-REALM FALSE POSITIVE. The guard added in #10256 used a bare
|
||||||
|
* `result.response instanceof Response`. OmniRoute's default egress
|
||||||
|
* (`open-sse/utils/proxyFetch.ts`) is the npm `undici` package's `fetch`,
|
||||||
|
* whose `Response` class is NOT `globalThis.Response` — so every ordinary
|
||||||
|
* upstream response arrived as a "contract violation". The guard must
|
||||||
|
* recognize a structurally valid Response from any realm.
|
||||||
|
*
|
||||||
|
* 2. TRANSIENT MISCLASSIFICATION. A genuine contract violation is an INTERNAL
|
||||||
|
* bug, not a flaky upstream. It carried no `.status`, so chatCore's default
|
||||||
|
* mapped it to 502 → the connection got cooled down as "rate limited", the
|
||||||
|
* provider breaker counted it, and `processSingleItemWithRetry` (which
|
||||||
|
* retries 429/502/504 up to 200×/24h) span forever. It must surface as a
|
||||||
|
* terminal internal 500 carrying a stable error code, and every resilience
|
||||||
|
* layer must treat that code as request-scoped: no cooldown, no breaker.
|
||||||
|
*/
|
||||||
|
import test from "node:test";
|
||||||
|
import assert from "node:assert/strict";
|
||||||
|
import { Response as UndiciResponse } from "undici";
|
||||||
|
|
||||||
|
import { normalizeExecutorResult } from "../../open-sse/handlers/chatCore/upstreamTimeouts.ts";
|
||||||
|
import { EXECUTOR_CONTRACT_VIOLATION_CODE } from "../../open-sse/config/constants.ts";
|
||||||
|
import {
|
||||||
|
isRequestScopedUpstreamFailure,
|
||||||
|
shouldSkipConnDisable,
|
||||||
|
} from "../../open-sse/services/combo/comboPredicates.ts";
|
||||||
|
import { shouldTripProviderBreakerForResult } from "../../src/sse/handlers/chatPredicates.ts";
|
||||||
|
import { checkFallbackError } from "../../open-sse/services/accountFallback.ts";
|
||||||
|
|
||||||
|
// ─── 1. Cross-realm Response acceptance ──────────────────────────────────────
|
||||||
|
|
||||||
|
test("undici's Response is a different class than the global one (premise)", () => {
|
||||||
|
assert.notEqual(
|
||||||
|
UndiciResponse as unknown,
|
||||||
|
globalThis.Response as unknown,
|
||||||
|
"if these ever become the same class the cross-realm guard below is moot"
|
||||||
|
);
|
||||||
|
assert.equal(
|
||||||
|
new UndiciResponse("x", { status: 200 }) instanceof globalThis.Response,
|
||||||
|
false,
|
||||||
|
"premise: an undici Response fails a bare `instanceof Response`"
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("normalizeExecutorResult accepts a cross-realm Response in the capture-object arm", () => {
|
||||||
|
const response = new UndiciResponse(JSON.stringify({ ok: true }), { status: 401 });
|
||||||
|
|
||||||
|
const normalized = normalizeExecutorResult({
|
||||||
|
response,
|
||||||
|
url: "https://api.openai.com/v1/chat/completions",
|
||||||
|
headers: { "x-req": "1" },
|
||||||
|
transformedBody: { a: 1 },
|
||||||
|
});
|
||||||
|
|
||||||
|
assert.equal(normalized.response, response as unknown);
|
||||||
|
assert.equal(normalized.response.status, 401);
|
||||||
|
assert.equal(normalized.url, "https://api.openai.com/v1/chat/completions");
|
||||||
|
assert.deepEqual(normalized.headers, { "x-req": "1" });
|
||||||
|
assert.deepEqual(normalized.transformedBody, { a: 1 });
|
||||||
|
});
|
||||||
|
|
||||||
|
test("normalizeExecutorResult accepts a bare cross-realm Response", () => {
|
||||||
|
const response = new UndiciResponse("body", { status: 503 });
|
||||||
|
|
||||||
|
const normalized = normalizeExecutorResult(response);
|
||||||
|
|
||||||
|
assert.equal(normalized.response, response as unknown);
|
||||||
|
assert.equal(normalized.response.status, 503);
|
||||||
|
assert.equal(normalized.url, "");
|
||||||
|
assert.deepEqual(normalized.headers, {});
|
||||||
|
assert.equal(normalized.transformedBody, null);
|
||||||
|
});
|
||||||
|
|
||||||
|
// ─── 2. A genuine violation is terminal, not a transient provider failure ────
|
||||||
|
|
||||||
|
function captureThrow(run: () => unknown): Error & { status?: unknown; code?: unknown } {
|
||||||
|
try {
|
||||||
|
run();
|
||||||
|
} catch (err) {
|
||||||
|
return err as Error & { status?: unknown; code?: unknown };
|
||||||
|
}
|
||||||
|
throw new assert.AssertionError({ message: "expected normalizeExecutorResult to throw" });
|
||||||
|
}
|
||||||
|
|
||||||
|
test("a genuinely malformed executor result still throws", () => {
|
||||||
|
assert.throws(() => normalizeExecutorResult({}), /must contain a Response/);
|
||||||
|
assert.throws(() => normalizeExecutorResult(undefined), /must contain a Response/);
|
||||||
|
assert.throws(() => normalizeExecutorResult({ response: "not-a-response" }), /must contain a/);
|
||||||
|
// A partial look-alike (no body readers) must NOT slip past the duck-type.
|
||||||
|
assert.throws(
|
||||||
|
() => normalizeExecutorResult({ response: { status: 200, ok: true } }),
|
||||||
|
/must contain a Response/
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("the contract-violation error carries an internal-terminal status + stable code", () => {
|
||||||
|
const err = captureThrow(() => normalizeExecutorResult({ response: "not-a-response" }));
|
||||||
|
|
||||||
|
assert.equal(err.status, 500, "an internal contract violation is a 500, never a provider 502");
|
||||||
|
assert.equal(
|
||||||
|
err.code,
|
||||||
|
EXECUTOR_CONTRACT_VIOLATION_CODE,
|
||||||
|
"chatCore reads `.code` (getUpstreamErrorIdentifier) to tag the surfaced error"
|
||||||
|
);
|
||||||
|
assert.equal(EXECUTOR_CONTRACT_VIOLATION_CODE, "executor_contract_violation");
|
||||||
|
});
|
||||||
|
|
||||||
|
test("the contract-violation code is classified as a request-scoped failure", () => {
|
||||||
|
assert.equal(isRequestScopedUpstreamFailure({ code: EXECUTOR_CONTRACT_VIOLATION_CODE }), true);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("a contract violation must not cool the connection down", () => {
|
||||||
|
assert.equal(
|
||||||
|
shouldSkipConnDisable(
|
||||||
|
{
|
||||||
|
status: 500,
|
||||||
|
errorCode: EXECUTOR_CONTRACT_VIOLATION_CODE,
|
||||||
|
errorType: null,
|
||||||
|
error: "Executor result must contain a Response",
|
||||||
|
},
|
||||||
|
false,
|
||||||
|
false,
|
||||||
|
"openai"
|
||||||
|
),
|
||||||
|
true,
|
||||||
|
"our own bug must never mark the operator's account as rate-limited/unavailable"
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("a contract violation must not trip the provider circuit breaker", () => {
|
||||||
|
assert.equal(
|
||||||
|
shouldTripProviderBreakerForResult(
|
||||||
|
{
|
||||||
|
status: 500,
|
||||||
|
errorCode: EXECUTOR_CONTRACT_VIOLATION_CODE,
|
||||||
|
errorType: null,
|
||||||
|
error: "Executor result must contain a Response",
|
||||||
|
},
|
||||||
|
false,
|
||||||
|
false
|
||||||
|
),
|
||||||
|
false,
|
||||||
|
"500 is a breaker-failure status, but this one never reached the provider"
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("checkFallbackError treats the contract violation as terminal — no retry, no cooldown", () => {
|
||||||
|
const decision = checkFallbackError(
|
||||||
|
500,
|
||||||
|
"[500]: Executor result must contain a Response",
|
||||||
|
0,
|
||||||
|
"gpt-4o-mini",
|
||||||
|
"openai",
|
||||||
|
null,
|
||||||
|
null,
|
||||||
|
{ code: EXECUTOR_CONTRACT_VIOLATION_CODE }
|
||||||
|
);
|
||||||
|
|
||||||
|
assert.equal(decision.shouldFallback, false, "retrying our own bug just reproduces it");
|
||||||
|
assert.equal(decision.cooldownMs, 0, "no connection cooldown for an internal defect");
|
||||||
|
assert.equal(decision.skipProviderBreaker, true);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("a real provider 500 is still retryable (the terminal branch is not over-broad)", () => {
|
||||||
|
const decision = checkFallbackError(500, "Internal server error", 0, null, "openai");
|
||||||
|
|
||||||
|
assert.equal(decision.shouldFallback, true);
|
||||||
|
assert.ok(decision.cooldownMs > 0, "a genuine upstream 500 keeps its backoff cooldown");
|
||||||
|
});
|
||||||
Reference in New Issue
Block a user