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
35 changes: 35 additions & 0 deletions open-sse/services/combo.ts
Original file line number Diff line number Diff line change
Expand Up @@ -158,6 +158,7 @@ import {
isStreamReadinessFailureErrorBody,
isStreamEarlyEofErrorBody,
isTokenLimitBreachErrorBody,
isLocalQueueCapacityErrorBody,
toRecordedTarget,
getExhaustedTargetSkipReason,
clampPercent,
Expand Down Expand Up @@ -1657,6 +1658,7 @@ export async function handleComboChat({
// FIX 5: a local per-API-key token-limit 429 must not cool shared accounts.
const isTokenLimitBreach =
result.status === 429 && isTokenLimitBreachErrorBody(errorBody);
const isLocalQueueCapacity = isLocalQueueCapacityErrorBody(errorBody);

// Fix #1681: Status 499 means client disconnected — stop combo loop immediately.
// There is no point trying fallback models when nobody is listening.
Expand All @@ -1675,6 +1677,22 @@ export async function handleComboChat({
// so the combo would wrongly fall through to the next model after a 499.
return { ok: false, response: result };
}
if (isLocalQueueCapacity) {
log.info(
"COMBO",
`Local rate-limit queue capacity reached for ${modelStr} — returning without upstream fallback`
);
recordComboRequest(combo.name, modelStr, {
success: false,
latencyMs: Date.now() - startTime,
fallbackCount,
strategy,
target: toRecordedTarget(target),
});
recordedAttempts++;
if (i > 0) fallbackCount++;
return { ok: false, response: result };
}

// Combo fallback is target-level orchestration: a non-ok target response is
// treated as local to that target and the combo continues to the next target.
Expand Down Expand Up @@ -2989,6 +3007,23 @@ async function handleRoundRobinCombo({

// FIX 5: a local per-API-key token-limit 429 must not cool shared accounts.
const isTokenLimitBreach = result.status === 429 && isTokenLimitBreachErrorBody(errorBody);
const isLocalQueueCapacity = isLocalQueueCapacityErrorBody(errorBody);

if (isLocalQueueCapacity) {
log.info(
"COMBO-RR",
`Local rate-limit queue capacity reached for ${modelStr} — returning without upstream fallback`
);
recordComboRequest(combo.name, modelStr, {
success: false,
latencyMs: Date.now() - startTime,
fallbackCount,
strategy: "round-robin",
target: toRecordedTarget(target),
});
recordedAttempts++;
return result;
}

// Round-robin uses the same target-level fallback rule as other combo
// strategies: non-ok target responses fall through to the next target.
Expand Down
25 changes: 24 additions & 1 deletion open-sse/services/combo/comboPredicates.ts
Original file line number Diff line number Diff line change
Expand Up @@ -197,6 +197,10 @@ const REQUEST_SCOPED_UPSTREAM_ERROR_CODES: Record<string, true> = {
upstream_response_failed: true,
// Local combo per-target timer (targetTimeoutRunner) — not a connection health signal.
combo_target_timeout: true,
// Local limiter queue-capacity codes — not a provider/connection health signal.
rate_limit_queue_timeout: true,
rate_limit_queue_full: true,
rate_limit_queue_wedged: true,
};

/** Request/model-specific failures must not poison provider-wide resilience state. */
Expand All @@ -206,7 +210,11 @@ export function isRequestScopedUpstreamFailure(error?: {
}): boolean {
const code = typeof error?.code === "string" ? error.code.toLowerCase() : "";
const type = typeof error?.type === "string" ? error.type.toLowerCase() : "";
return REQUEST_SCOPED_UPSTREAM_ERROR_CODES[code] === true || type === "context_length_exceeded";
return (
REQUEST_SCOPED_UPSTREAM_ERROR_CODES[code] === true ||
type === "context_length_exceeded" ||
type === "local_queue_capacity"
);
}

/** Request-scoped classification that also has access to the HTTP body. */
Expand Down Expand Up @@ -358,6 +366,21 @@ export function isTokenLimitBreachErrorBody(errorBody: unknown): boolean {
return (error as Record<string, unknown>).code === "TOKEN_LIMIT_EXCEEDED";
}

/** Local limiter capacity is not an upstream/provider failure and must not cascade. */
export function isLocalQueueCapacityErrorBody(errorBody: unknown): boolean {
if (!errorBody || typeof errorBody !== "object") return false;
const error = (errorBody as Record<string, unknown>).error;
if (!error || typeof error !== "object") return false;
const code = String((error as Record<string, unknown>).code || "").toUpperCase();
const type = String((error as Record<string, unknown>).type || "").toLowerCase();
return (
code === "RATE_LIMIT_QUEUE_TIMEOUT" ||
code === "RATE_LIMIT_QUEUE_FULL" ||
code === "RATE_LIMIT_QUEUE_WEDGED" ||
type === "local_queue_capacity"
);
}

export function toRecordedTarget(target: ResolvedComboTarget) {
return {
executionKey: target.executionKey,
Expand Down
79 changes: 79 additions & 0 deletions tests/unit/rate-limit-local-capacity-classification.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,79 @@
import assert from "node:assert/strict";
import test from "node:test";

const { isLocalQueueCapacityErrorBody, isRequestScopedUpstreamFailure, shouldSkipConnDisable } =
await import("../../open-sse/services/combo/comboPredicates.ts");
const { handleComboChat } = await import("../../open-sse/services/combo.ts");

function localQueueResponse() {
return new Response(
JSON.stringify({
error: {
message: "Local rate-limit queue wait exceeded",
code: "RATE_LIMIT_QUEUE_TIMEOUT",
type: "local_queue_capacity",
},
}),
{ status: 429, headers: { "content-type": "application/json" } }
);
}

function createLog() {
return { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} };
}

test("local queue capacity is request-scoped and never disables a healthy connection", () => {
assert.equal(
isRequestScopedUpstreamFailure({
code: "RATE_LIMIT_QUEUE_TIMEOUT",
type: "local_queue_capacity",
}),
true
);
assert.equal(
shouldSkipConnDisable(
{
status: 429,
errorCode: "RATE_LIMIT_QUEUE_TIMEOUT",
errorType: "local_queue_capacity",
},
false,
false,
"nvidia"
),
true
);
assert.equal(
isLocalQueueCapacityErrorBody({
error: { code: "RATE_LIMIT_QUEUE_TIMEOUT", type: "local_queue_capacity" },
}),
true
);
});

test("combo returns a local queue capacity response without retrying or rotating targets", async () => {
let calls = 0;
const response = await handleComboChat({
body: { model: "openai/gpt-4" },
combo: {
name: `local-queue-capacity-${Math.random().toString(16).slice(2)}`,
strategy: "priority",
models: ["openai/gpt-4", "openai/gpt-4o-mini"],
config: { maxRetries: 2, maxSetRetries: 1, retryDelayMs: 0, fallbackDelayMs: 0 },
},
handleSingleModel: async () => {
calls += 1;
return localQueueResponse();
},
isModelAvailable: async () => true,
log: createLog() as never,
settings: {},
allCombos: null,
});

assert.equal(response.status, 429);
assert.equal(calls, 1, "a local queue limit must not amplify into retries or fallback calls");
const body = await response.json();
assert.equal(body.error.code, "RATE_LIMIT_QUEUE_TIMEOUT");
assert.equal(body.error.type, "local_queue_capacity");
});
Loading