diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 894c79033c..2b09178737 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -926,6 +926,9 @@ async function handleComboChatInner({ if (activeNativeTurnPin) { orderedTargets = applyNativeCodexTurnPin(orderedTargets, activeNativeTurnPin); if (orderedTargets.length === 0) { + // #11371: quota-share ordering already reserved a winner slot; release it on + // this early exit (idempotent). + targetResolution.quotaShareRelease?.(); return errorResponse( 409, "The pinned native Codex turn target is no longer available; the turn cannot be moved to another provider" @@ -952,7 +955,8 @@ async function handleComboChatInner({ // Surface a recovery hint + auto-clear the session pin after enough consecutive // no-target failures (silent-stop fix). Threshold of 3 prevents a one-off account // wipe from destroying the prompt-cache pin benefit on the next request. - recordComboFailure(effectiveSessionId, combo.name); + // #11371: same early-exit release as the pinned-turn path above. + targetResolution.quotaShareRelease?.(); return errorResponseWithComboDiagnostics( 404, "Combo has no executable targets", @@ -2841,6 +2845,9 @@ async function handleComboChatInner({ return await dispatchWithCooldownRetry(); } finally { quotaShareConcurrencyRelease?.(); + // #11371: release the in-flight slot quota-share ordering reserved for its + // winner — the counter must not leak monotonically upward across requests. + targetResolution.quotaShareRelease?.(); // G2: Clean up candidate registry to prevent unbounded memory growth. _unregisterExecutionCandidates(_registeredExecutionKeys); } diff --git a/open-sse/services/combo/applyStrategyOrdering.ts b/open-sse/services/combo/applyStrategyOrdering.ts index 38f07fbcdc..e077082c5e 100644 --- a/open-sse/services/combo/applyStrategyOrdering.ts +++ b/open-sse/services/combo/applyStrategyOrdering.ts @@ -20,6 +20,21 @@ import { } from "./targetSorters.ts"; import type { ComboLike, ComboLogger, ResolvedComboTarget } from "./types.ts"; +/** + * Result of {@link applyStrategyOrdering}. + * + * `quotaShareRelease` carries the idempotent release for the in-flight slot that + * quota-share ordering reserves for its winner (#11371). It is non-null only when + * the `quota-share` strategy ran; every other strategy leaves it null. The caller + * MUST invoke it exactly once when the request settles — selection reserves the + * slot, so dropping the callback leaks the counter monotonically upward and + * degenerates P2C into "fewest lifetime dispatches". + */ +export interface ApplyStrategyOrderingResult { + orderedTargets: ResolvedComboTarget[]; + quotaShareRelease: (() => void) | null; +} + export interface ApplyStrategyOrderingDeps { combo: ComboLike; config: Record; @@ -45,9 +60,10 @@ export async function applyStrategyOrdering( strategy: string, initialOrderedTargets: ResolvedComboTarget[], deps: ApplyStrategyOrderingDeps -): Promise { +): Promise { const { combo, config, body, log, apiKeyAllowedConnections, sessionKey } = deps; let orderedTargets = initialOrderedTargets; + let quotaShareRelease: (() => void) | null = null; if (strategy === "lkgp") { try { @@ -229,14 +245,18 @@ export async function applyStrategyOrdering( const qsModel = typeof body?.model === "string" ? body.model : (orderedTargets[0]?.modelStr ?? ""); const qsMaxConcurrent = await resolveMaxConcurrentByConnection(orderedTargets); - orderedTargets = selectQuotaShareTarget(orderedTargets, combo.name, qsModel, Date.now(), { + const qsSelection = selectQuotaShareTarget(orderedTargets, combo.name, qsModel, Date.now(), { maxConcurrentByConnection: qsMaxConcurrent, - }).orderedTargets; + }); + orderedTargets = qsSelection.orderedTargets; + // #11371: the reservation made inside selectQuotaShareTarget must outlive this + // call — hand the release to the host so it can fire it when the request settles. + quotaShareRelease = qsSelection.decrementInflight; log.info( "COMBO", `Quota-share ordering: ${orderedTargets[0]?.modelStr}${orderedTargets[0]?.connectionId ? ` (${orderedTargets[0].connectionId})` : ""} selected (DRR+P2C)` ); } - return orderedTargets; + return { orderedTargets, quotaShareRelease }; } diff --git a/open-sse/services/combo/targetResolution.ts b/open-sse/services/combo/targetResolution.ts index eeec177c86..5022f2caa7 100644 --- a/open-sse/services/combo/targetResolution.ts +++ b/open-sse/services/combo/targetResolution.ts @@ -122,6 +122,13 @@ export interface ResolvedComboTargetPipeline { /** Session-stickiness result — the attempt loop reads `.messageHash` on success/failure. */ sticky: ApplyStickinessResult; preScreenMap: Map; + /** + * Idempotent release for the in-flight slot quota-share ordering reserved for + * its winner (#11371). Null unless the `quota-share` strategy ran. The host MUST + * invoke it when the request settles; this pipeline already releases it on any + * earlyResponse it produces after selection. + */ + quotaShareRelease: (() => void) | null; } export type ResolveComboTargetPipelineResult = @@ -394,7 +401,11 @@ async function orderByStrategy( initialOrderedTargets: ResolvedComboTarget[] ): Promise< | { earlyResponse: Response } - | { orderedTargets: ResolvedComboTarget[]; autoUsedExplicitRouter: boolean } + | { + orderedTargets: ResolvedComboTarget[]; + autoUsedExplicitRouter: boolean; + quotaShareRelease: (() => void) | null; + } > { const { strategy, body, combo, settings, config, log } = deps; if (strategy === "auto") { @@ -413,17 +424,22 @@ async function orderByStrategy( return { orderedTargets: autoResult.orderedTargets, autoUsedExplicitRouter: autoResult.autoUsedExplicitRouter, + quotaShareRelease: null, }; } - const orderedTargets = await applyStrategyOrdering(strategy, initialOrderedTargets, { - combo, - config, - body, - log, - apiKeyAllowedConnections: deps.apiKeyAllowedConnections, - sessionKey: deps.relayOptions?.sessionId, - }); - return { orderedTargets, autoUsedExplicitRouter: false }; + const { orderedTargets, quotaShareRelease } = await applyStrategyOrdering( + strategy, + initialOrderedTargets, + { + combo, + config, + body, + log, + apiKeyAllowedConnections: deps.apiKeyAllowedConnections, + sessionKey: deps.relayOptions?.sessionId, + } + ); + return { orderedTargets, autoUsedExplicitRouter: false, quotaShareRelease }; } /** @@ -714,10 +730,15 @@ export async function resolveComboTargetPipeline( const ordering = await orderByStrategy(deps, orderedTargets); if ("earlyResponse" in ordering) return ordering; - const { autoUsedExplicitRouter } = ordering; + const { autoUsedExplicitRouter, quotaShareRelease } = ordering; const continuity = await applyContinuityFilters(deps, ordering.orderedTargets); - if ("earlyResponse" in continuity) return continuity; + if ("earlyResponse" in continuity) { + // #11371: selection already reserved the winner's in-flight slot; a hard + // filter exhausting the pool must not leak it. + quotaShareRelease?.(); + return continuity; + } orderedTargets = applyTaskAwareOrdering(deps, continuity.orderedTargets, autoUsedExplicitRouter); orderedTargets = await applyPromptCacheStage( deps, @@ -741,5 +762,6 @@ export async function resolveComboTargetPipeline( getWeightedStepKeyForTarget, sticky: continuity.sticky, preScreenMap, + quotaShareRelease, }; } diff --git a/tests/unit/combo-apply-strategy-ordering-split.test.ts b/tests/unit/combo-apply-strategy-ordering-split.test.ts index 12eebb96f6..90547c29a9 100644 --- a/tests/unit/combo-apply-strategy-ordering-split.test.ts +++ b/tests/unit/combo-apply-strategy-ordering-split.test.ts @@ -50,7 +50,7 @@ test("unknown strategy -> input order unchanged (same reference contents)", asyn const input = [target("openai", "gpt-4o"), target("anthropic", "claude-3")]; const out = await applyStrategyOrdering("no-such-strategy", input, deps()); assert.deepEqual( - out.map((t: { executionKey: string }) => t.executionKey), + out.orderedTargets.map((t: { executionKey: string }) => t.executionKey), ["openai>gpt-4o", "anthropic>claude-3"] ); }); @@ -59,7 +59,7 @@ test("fill-first -> preserves priority order", async () => { const input = [target("a", "m1"), target("b", "m2"), target("c", "m3")]; const out = await applyStrategyOrdering("fill-first", input, deps()); assert.deepEqual( - out.map((t: { executionKey: string }) => t.executionKey), + out.orderedTargets.map((t: { executionKey: string }) => t.executionKey), ["a>m1", "b>m2", "c>m3"] ); }); @@ -67,8 +67,8 @@ test("fill-first -> preserves priority order", async () => { test("random -> same multiset of targets (a permutation)", async () => { const input = [target("a", "m1"), target("b", "m2"), target("c", "m3")]; const out = await applyStrategyOrdering("random", input, deps()); - assert.equal(out.length, 3); - assert.deepEqual(keys(out), keys(input)); + assert.equal(out.orderedTargets.length, 3); + assert.deepEqual(keys(out.orderedTargets), keys(input)); }); test("cost-optimized manifest routing logs through the canonical strategy path", async () => { @@ -90,7 +90,7 @@ test("cost-optimized manifest routing logs through the canonical strategy path", log, } as never); - assert.equal(out.length, 2); + assert.equal(out.orderedTargets.length, 2); assert.equal( debugCalls.some((args) => args[1] === "manifest routing applied"), true, diff --git a/tests/unit/prompt-cache-affinity.test.ts b/tests/unit/prompt-cache-affinity.test.ts index 7e25456f02..eb980700f8 100644 --- a/tests/unit/prompt-cache-affinity.test.ts +++ b/tests/unit/prompt-cache-affinity.test.ts @@ -134,7 +134,7 @@ test("cache-optimized strategy routes a stable prompt key to the same account", }; const first = await applyStrategyOrdering("cache-optimized", targets, deps); const second = await applyStrategyOrdering("cache-optimized", [...targets].reverse(), deps); - assert.equal(first[0].connectionId, second[0].connectionId); + assert.equal(first.orderedTargets[0].connectionId, second.orderedTargets[0].connectionId); }); test("foreign OAuth session softly redirects cache affinity while the same session stays local", () => { @@ -145,11 +145,18 @@ test("foreign OAuth session softly redirects cache affinity while the same sessi ]; const fixture = Array.from({ length: 10_000 }, (_, index) => { const body = { prompt_cache_key: `occupied-cache-key-${index}` }; - const baseline = applyPromptCacheAffinity(oauthTargets, body, true, "global", "session-a").targets; + const baseline = applyPromptCacheAffinity( + oauthTargets, + body, + true, + "global", + "session-a" + ).targets; const occupied = baseline[0]; const alternative = baseline[1]; const release = reserveOAuthSession(occupied.connectionId!, "session-a"); - const foreignFirst = applyPromptCacheAffinity(oauthTargets, body, true, "global", "session-b").targets[0]; + const foreignFirst = applyPromptCacheAffinity(oauthTargets, body, true, "global", "session-b") + .targets[0]; release(); return foreignFirst.connectionId === alternative.connectionId ? { body, occupied, alternative } @@ -160,12 +167,14 @@ test("foreign OAuth session softly redirects cache affinity while the same sessi const release = reserveOAuthSession(occupied.connectionId!, "session-a"); assert.equal( - applyPromptCacheAffinity(oauthTargets, body, true, "global", "session-a").targets[0].connectionId, + applyPromptCacheAffinity(oauthTargets, body, true, "global", "session-a").targets[0] + .connectionId, occupied.connectionId, "the owning session keeps its cache-local account" ); assert.equal( - applyPromptCacheAffinity(oauthTargets, body, true, "global", "session-b").targets[0].connectionId, + applyPromptCacheAffinity(oauthTargets, body, true, "global", "session-b").targets[0] + .connectionId, alternative.connectionId, "a foreign session prefers the free OAuth account" ); diff --git a/tests/unit/quota-share-strategy.test.ts b/tests/unit/quota-share-strategy.test.ts index 7d1e59f940..99d5c0fcbb 100644 --- a/tests/unit/quota-share-strategy.test.ts +++ b/tests/unit/quota-share-strategy.test.ts @@ -21,6 +21,7 @@ import { _clearDrrStateForTest, _getDrrDeficitForTest, } from "../../open-sse/services/combo/quotaShareStrategy.ts"; +import { applyStrategyOrdering } from "../../open-sse/services/combo/applyStrategyOrdering.ts"; import { incrementInflight, decrementInflight, @@ -490,3 +491,48 @@ describe("activation: qtSd/ combos use strategy 'quota-share'", () => { ); }); }); + +// ─── #11371: release must travel out of the ordering path ─────────────────── + +describe("applyStrategyOrdering threads the in-flight release (#11371)", () => { + const noopLog = { info() {}, warn() {}, error() {}, debug() {} }; + + test("quota-share ordering returns a release that restores the counter to 0", async () => { + const targets = [makeTarget("ek-rel-a", "conn-rel-a"), makeTarget("ek-rel-b", "conn-rel-b")]; + const { orderedTargets, quotaShareRelease } = await applyStrategyOrdering( + "quota-share", + targets, + { + combo: { id: "c-rel", name: "qtSd/rel" }, + config: {}, + body: { model: "anthropic/claude-sonnet-4-5" }, + log: noopLog, + apiKeyAllowedConnections: null, + } as never + ); + + assert.ok(orderedTargets.length > 0); + assert.ok(quotaShareRelease, "quota-share ordering must hand back a release callback"); + const winnerConn = orderedTargets[0].connectionId ?? ""; + assert.ok(winnerConn, "winner must carry a connectionId for the reservation"); + // Selection reserved exactly one slot on the winner's connection. + assert.equal(getInflight(winnerConn, NOW), 1, "selection must reserve one in-flight slot"); + // The real-caller contract: release once when the request settles. + quotaShareRelease!(); + assert.equal(getInflight(winnerConn, NOW), 0, "release must return the counter to 0"); + // Idempotent: a second call must not push the counter negative. + quotaShareRelease!(); + assert.equal(getInflight(winnerConn, NOW), 0, "double release floors at 0"); + }); + + test("non-quota-share strategies leave quotaShareRelease null", async () => { + const result = await applyStrategyOrdering("fill-first", [makeTarget("ek-null", "conn-null")], { + combo: { id: "c-ff", name: "plain" }, + config: {}, + body: {}, + log: noopLog, + apiKeyAllowedConnections: null, + } as never); + assert.equal(result.quotaShareRelease, null); + }); +});