diff --git a/open-sse/services/__tests__/fail-fast-concurrency-gate.test.ts b/open-sse/services/__tests__/fail-fast-concurrency-gate.test.ts new file mode 100644 index 0000000000..cd969fb352 --- /dev/null +++ b/open-sse/services/__tests__/fail-fast-concurrency-gate.test.ts @@ -0,0 +1,30 @@ +import { describe, it, expect, beforeEach } from "vitest"; +import { acquire, isAccountSemaphoreFull, resetAll } from "../accountSemaphore.ts"; + +describe("isAccountSemaphoreFull fail-fast concurrency gate", () => { + beforeEach(() => { + resetAll(); + }); + + it("returns false when no semaphore gate exists", () => { + expect(isAccountSemaphoreFull("featherless-ai", "conn-1", 1)).toBe(false); + }); + + it("returns false when maxConcurrency is null, <= 0, or bypassed", () => { + expect(isAccountSemaphoreFull("featherless-ai", "conn-1", null)).toBe(false); + expect(isAccountSemaphoreFull("featherless-ai", "conn-1", 0)).toBe(false); + }); + + it("returns false when running < maxConcurrency", async () => { + const release = await acquire("featherless-ai:conn-1", { maxConcurrency: 2 }); + expect(isAccountSemaphoreFull("featherless-ai", "conn-1", 2)).toBe(false); + release(); + }); + + it("returns true immediately when running >= maxConcurrency", async () => { + const release = await acquire("featherless-ai:conn-1", { maxConcurrency: 1 }); + expect(isAccountSemaphoreFull("featherless-ai", "conn-1", 1)).toBe(true); + release(); + expect(isAccountSemaphoreFull("featherless-ai", "conn-1", 1)).toBe(false); + }); +}); diff --git a/open-sse/services/accountFallback.ts b/open-sse/services/accountFallback.ts index be048f69c9..b51cace449 100644 --- a/open-sse/services/accountFallback.ts +++ b/open-sse/services/accountFallback.ts @@ -1631,7 +1631,11 @@ export function checkFallbackError( !errorStr.toLowerCase().includes("hour quota") && !errorStr.toLowerCase().includes("quota has been exceeded") ) { - return resolveApiKeyForbiddenFallback(errorStr, buildRetryableFallback, RateLimitReason.AUTH_ERROR); + return resolveApiKeyForbiddenFallback( + errorStr, + buildRetryableFallback, + RateLimitReason.AUTH_ERROR + ); } } @@ -1953,6 +1957,8 @@ export function applyErrorState( return nextState; } +export { isAccountSemaphoreFull } from "./accountSemaphore.ts"; + /** * Get account health score (0-100) for P2C selection (Phase 9) * @param {object} account diff --git a/open-sse/services/accountSemaphore.ts b/open-sse/services/accountSemaphore.ts index affb06a41f..ddb629e12e 100644 --- a/open-sse/services/accountSemaphore.ts +++ b/open-sse/services/accountSemaphore.ts @@ -342,6 +342,24 @@ export function getStats(): Record { return stats; } +/** + * Check if an account semaphore key is currently at or over its max concurrency limit. + * Returns true if running >= maxConcurrency or blocked. + */ +export function isAccountSemaphoreFull( + provider: string, + accountKey: string, + maxConcurrency?: number | null +): boolean { + if (isBypassed(maxConcurrency)) return false; + const key = buildAccountSemaphoreKey({ provider, accountKey }); + const gate = gates.get(key); + if (!gate) return false; + const effectiveCap = maxConcurrency ?? gate.maxConcurrency; + if (isBypassed(effectiveCap)) return false; + return gate.running >= effectiveCap || isBlocked(gate); +} + /** * Reset a single key and reject queued waiters. */ diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index acfb0e81c6..37c270e25c 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -14,6 +14,7 @@ import { getModelLockoutInfo, getRuntimeProviderProfile, hasPerModelQuota, + isAccountSemaphoreFull, isModelLocked, MODEL_ACCESS_DENIED_PATTERNS, recordModelLockoutFailure, @@ -1012,6 +1013,20 @@ export async function handleComboChat({ if (i > 0) fallbackCount++; return null; } + + // Concurrency gate: fail-fast skip when connection is at max_concurrent capacity (e.g. Featherless 1/1) + const maxConcurrentCap = await lookupPositiveCap(connectionId); + if ( + maxConcurrentCap && + isAccountSemaphoreFull(provider, connectionId, maxConcurrentCap) + ) { + log.info( + "COMBO", + `Skipping ${modelStr} — connection ${connectionId} is at max concurrency cap (${maxConcurrentCap})` + ); + if (i > 0) fallbackCount++; + return null; + } } // Retry loop for transient errors diff --git a/open-sse/services/combo/runtimeUnitCapacity.ts b/open-sse/services/combo/runtimeUnitCapacity.ts new file mode 100644 index 0000000000..4d0265b7a8 --- /dev/null +++ b/open-sse/services/combo/runtimeUnitCapacity.ts @@ -0,0 +1,80 @@ +/** + * @file runtimeUnitCapacity.ts + * @description Concurrency-capacity checks for nested combo execute-mode units so + * ordered strategies overflow to the next slot instead of queueing on a full connection. + * + * @changes + * - [2026-07-24] [Composer] - Initial capacity pre-check for execute-mode runtime units + */ +import { isAccountSemaphoreFull } from "../accountSemaphore.ts"; +import { resolveComboTargets } from "./comboStructure.ts"; +import { lookupPositiveCap } from "./concurrencyCaps.ts"; +import type { ComboCollectionLike, ComboLike, ResolvedComboUnit } from "./types.ts"; + +type CapLookup = (connectionId: string) => Promise; + +function isRecord(value: unknown): value is Record { + return !!value && typeof value === "object" && !Array.isArray(value); +} + +function getCombosList(allCombos: ComboCollectionLike): ComboLike[] { + const combos = Array.isArray(allCombos) ? allCombos : allCombos?.combos || []; + return combos.filter( + (combo): combo is ComboLike => isRecord(combo) && typeof combo.name === "string" + ); +} + +function findComboByName(allCombos: ComboCollectionLike, name: string): ComboLike | null { + return getCombosList(allCombos).find((combo) => combo.name === name) || null; +} + +async function isConnectionAtConcurrencyCap( + provider: string, + connectionId: string, + lookupCap: CapLookup +): Promise { + const cap = await lookupCap(connectionId); + if (!cap) return false; + return isAccountSemaphoreFull(provider, connectionId, cap); +} + +/** + * Returns true when the runtime unit should be skipped because every limited + * connection it would use is already at max_concurrent. + */ +export async function isRuntimeUnitAtConcurrencyCap( + unit: ResolvedComboUnit, + allCombos: ComboCollectionLike, + lookupCap: CapLookup = lookupPositiveCap +): Promise { + if (unit.kind === "model") { + if (!unit.connectionId || !unit.provider) return false; + return isConnectionAtConcurrencyCap(unit.provider, unit.connectionId, lookupCap); + } + + const childCombo = findComboByName(allCombos, unit.comboName); + if (!childCombo) return false; + + const targets = resolveComboTargets(childCombo, allCombos, 1); + const byConnection = new Map(); + for (const target of targets) { + if (!target.connectionId || !target.provider) continue; + byConnection.set(target.connectionId, { + provider: target.provider, + connectionId: target.connectionId, + }); + } + if (byConnection.size === 0) return false; + + let sawLimitedConnection = false; + for (const { provider, connectionId } of byConnection.values()) { + const cap = await lookupCap(connectionId); + if (!cap) continue; + sawLimitedConnection = true; + if (!isAccountSemaphoreFull(provider, connectionId, cap)) { + return false; + } + } + + return sawLimitedConnection; +} diff --git a/open-sse/services/combo/runtimeUnits.ts b/open-sse/services/combo/runtimeUnits.ts index e839fc01bb..28128d11f3 100644 --- a/open-sse/services/combo/runtimeUnits.ts +++ b/open-sse/services/combo/runtimeUnits.ts @@ -1,7 +1,14 @@ -// Nested combo runtime unit execution — see combo.ts for integration. +/** + * @file runtimeUnits.ts + * @description Nested combo runtime unit execution — see combo.ts for integration. + * + * @changes + * - [2026-07-24] [Composer] - Skip execute-mode units at concurrency cap before dispatch + */ import { errorResponse } from "../../utils/error.ts"; import { recordComboRequest } from "../comboMetrics.ts"; import { resolveDelayMs } from "./comboPredicates.ts"; +import { isRuntimeUnitAtConcurrencyCap } from "./runtimeUnitCapacity.ts"; import { validateResponseQuality, releaseQualityClone } from "./validateQuality.ts"; import type { ResponseValidationConfig } from "./responseValidation.ts"; import type { @@ -190,6 +197,15 @@ export async function executeRuntimeUnitCombo(args: { let fallbackCount = 0; for (const unit of orderedUnits) { + if (await isRuntimeUnitAtConcurrencyCap(unit, args.allCombos)) { + args.log.info( + "COMBO", + `Skipping ${unit.kind} ${unitDisplayName(unit)} — concurrency cap reached` + ); + fallbackCount += 1; + continue; + } + for (let retry = 0; retry <= maxRetries; retry += 1) { if (args.signal?.aborted) return { response: errorResponse(499, "Client disconnected"), unit }; diff --git a/tests/unit/combo-runtime-unit-concurrency.test.ts b/tests/unit/combo-runtime-unit-concurrency.test.ts new file mode 100644 index 0000000000..d496bb7c2a --- /dev/null +++ b/tests/unit/combo-runtime-unit-concurrency.test.ts @@ -0,0 +1,145 @@ +/** + * @file combo-runtime-unit-concurrency.test.ts + * @description Regression tests for execute-mode concurrency overflow (skip full units). + * + * @changes + * - [2026-07-24] [Composer] - Initial execute-mode capacity overflow coverage + */ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { + acquire, + buildAccountSemaphoreKey, + resetAll, +} from "../../open-sse/services/accountSemaphore.ts"; +import { isRuntimeUnitAtConcurrencyCap } from "../../open-sse/services/combo/runtimeUnitCapacity.ts"; +import type { ResolvedComboUnit } from "../../open-sse/services/combo/types.ts"; + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-runtime-unit-cap-")); +const ORIGINAL_DATA_DIR = process.env.DATA_DIR; +process.env.DATA_DIR = TEST_DATA_DIR; + +const { handleComboChat } = await import("../../open-sse/services/combo.ts"); +const { getDbInstance, resetDbInstance } = await import("../../src/lib/db/core.ts"); + +function createLog() { + return { + info: () => {}, + warn: () => {}, + error: () => {}, + debug: () => {}, + }; +} + +function okResponse() { + return new Response(JSON.stringify({ choices: [{ message: { content: "ok" } }] }), { + status: 200, + headers: { "content-type": "application/json" }, + }); +} + +function seedConnection(id: string, provider: string, maxConcurrent: number) { + const db = getDbInstance(); + db.prepare( + `INSERT INTO provider_connections + (id, provider, auth_type, is_active, max_concurrent, created_at, updated_at) + VALUES (?, ?, 'apikey', 1, ?, datetime('now'), datetime('now'))` + ).run(id, provider, maxConcurrent); +} + +test.after(() => { + resetAll(); + resetDbInstance(); + if (ORIGINAL_DATA_DIR === undefined) delete process.env.DATA_DIR; + else process.env.DATA_DIR = ORIGINAL_DATA_DIR; + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); +}); + +test("isRuntimeUnitAtConcurrencyCap returns true for a model unit at cap", async () => { + const connectionId = "conn-feather-cap"; + const provider = "featherless-ai"; + const key = buildAccountSemaphoreKey({ provider, accountKey: connectionId }); + const release1 = await acquire(key, { maxConcurrency: 2 }); + const release2 = await acquire(key, { maxConcurrency: 2 }); + + const unit: ResolvedComboUnit = { + kind: "model", + stepId: "step-1", + executionKey: "step-1", + modelStr: "featherless-ai/deepseek-ai/DeepSeek-V4-Pro", + provider, + providerId: provider, + connectionId, + weight: 0, + label: null, + }; + + assert.equal( + await isRuntimeUnitAtConcurrencyCap(unit, [], async () => 2), + true, + "unit should be at cap when two slots are in use" + ); + + release1(); + assert.equal( + await isRuntimeUnitAtConcurrencyCap(unit, [], async () => 2), + false, + "unit should have headroom after one slot frees" + ); + release2(); +}); + +test("execute fill-first overflows to the next unit when the first connection is at cap", async () => { + const primaryConnectionId = "conn-primary-overflow"; + const backupConnectionId = "conn-backup-overflow"; + const provider = "featherless-ai"; + seedConnection(primaryConnectionId, provider, 2); + seedConnection(backupConnectionId, "alibaba", 2); + + const key = buildAccountSemaphoreKey({ provider, accountKey: primaryConnectionId }); + const release1 = await acquire(key, { maxConcurrency: 2 }); + const release2 = await acquire(key, { maxConcurrency: 2 }); + + const calls: string[] = []; + const combo = { + name: "overflow-execute", + strategy: "fill-first", + models: [ + { + kind: "model", + model: "featherless-ai/deepseek-ai/DeepSeek-V4-Pro", + providerId: provider, + connectionId: primaryConnectionId, + }, + { + kind: "model", + model: "alibaba/qwen3.7-max-preview", + providerId: "alibaba", + connectionId: backupConnectionId, + }, + ], + config: { nestedComboMode: "execute", maxRetries: 0, retryDelayMs: 0 }, + }; + + const result = await handleComboChat({ + body: {}, + combo, + handleSingleModel: async (_body, modelStr) => { + calls.push(modelStr); + return okResponse(); + }, + isModelAvailable: async () => true, + log: createLog(), + settings: null, + allCombos: [combo], + }); + + assert.equal(result.ok, true); + assert.deepEqual(calls, ["alibaba/qwen3.7-max-preview"]); + + release1(); + release2(); +});