diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index a2bc7ea5bd85..a2f372b1d4ea 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -27,7 +27,7 @@ export { relocateDirectiveOnlyMessages, } from "./chatCore/claudeSystemRole.ts"; import { checkIdempotencyCache } from "./chatCore/idempotency.ts"; -import { acquireTurnExecution, createTurnInProgressResult } from "./chatCore/turnExecutionGuard.ts"; +import { acquireTurnExecutionWithWait, createTurnInProgressResult } from "./chatCore/turnExecutionGuard.ts"; import { checkSemanticCache, isSemanticCacheEnabled } from "./chatCore/semanticCache.ts"; import { checkLifecycle, resolveLifecycle } from "./chatCore/modelLifecyclePolicy.ts"; import { @@ -580,7 +580,7 @@ async function handleChatCoreInner({ return idempotencyHit; } - const turnExecution = acquireTurnExecution(idempotencyKey); + const turnExecution = await acquireTurnExecutionWithWait(idempotencyKey, { log, traceId }); if (turnExecution.acquired === false) { const duplicate = createTurnInProgressResult(turnExecution.retryCount); log?.warn?.( diff --git a/open-sse/handlers/chatCore/turnExecutionGuard.ts b/open-sse/handlers/chatCore/turnExecutionGuard.ts index d2adc1d61c2c..fe48c35b3726 100644 --- a/open-sse/handlers/chatCore/turnExecutionGuard.ts +++ b/open-sse/handlers/chatCore/turnExecutionGuard.ts @@ -75,6 +75,57 @@ export function getTurnExecutionSnapshot(key: string) { return { retryCount: active.retryCount, ageMs: Math.max(0, Date.now() - active.startedAt) }; } +export const TURN_WAIT_TIMEOUT_MS = 60_000; +export const TURN_WAIT_POLL_MS = 500; + +export type TurnExecutionWaitOptions = { + /** Upper bound on how long a duplicate waits for the in-flight turn to release. */ + timeoutMs?: number; + /** Poll interval while waiting. */ + pollMs?: number; + /** Client abort signal — a disconnected caller stops waiting immediately. */ + signal?: AbortSignal | null; + /** Warn-channel logger receiving the duplicate-waiting notice. */ + log?: { warn?: (tag: string, msg: string) => void } | null | undefined; + traceId?: string; +}; + +/** + * Acquires the turn slot, waiting out an in-flight duplicate instead of + * failing it immediately. + * + * Executing a duplicate's request path straight to a 409 turns a transient + * self-collision — executor/account-fallback retries re-entering chatCore + * with the caller's headers while the original turn is stuck on a slow + * upstream — into a fatal client-visible failure, and the 409 is then + * misread downstream as an account-level rate limit ("reset after Ns"), + * cooling down healthy accounts. Waiting keeps the guard's serialization + * guarantee without failing the duplicate; after the timeout the caller + * falls back to the 409 result as before. + */ +export async function acquireTurnExecutionWithWait( + key: string | null, + opts: TurnExecutionWaitOptions = {} +): Promise { + const turn = acquireTurnExecution(key); + if (turn.acquired) return turn; + + const timeoutMs = opts.timeoutMs ?? TURN_WAIT_TIMEOUT_MS; + const pollMs = opts.pollMs ?? TURN_WAIT_POLL_MS; + const { signal, log, traceId } = opts; + log?.warn?.( + "TURN_GUARD", + `duplicate waiting cid=${traceId ?? "?"} retry=${turn.retryCount} ageMs=${turn.ageMs}` + ); + const deadline = Date.now() + timeoutMs; + let current = turn; + while (!current.acquired && Date.now() < deadline && !signal?.aborted) { + await new Promise((resolve) => setTimeout(resolve, pollMs)); + current = acquireTurnExecution(key); + } + return current; +} + export function clearTurnExecutionsForTesting(): void { activeTurns.clear(); } diff --git a/tests/unit/chatcore/turn-execution-guard-wait.test.ts b/tests/unit/chatcore/turn-execution-guard-wait.test.ts new file mode 100644 index 000000000000..058f78b9927a --- /dev/null +++ b/tests/unit/chatcore/turn-execution-guard-wait.test.ts @@ -0,0 +1,62 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { + acquireTurnExecution, + acquireTurnExecutionWithWait, + clearTurnExecutionsForTesting, +} from "../../../open-sse/handlers/chatCore/turnExecutionGuard.ts"; + +test("acquires immediately when no turn is in flight", async () => { + clearTurnExecutionsForTesting(); + const turn = await acquireTurnExecutionWithWait("k"); + assert.equal(turn.acquired, true); +}); + +test("null key remains an unconditional acquire", async () => { + clearTurnExecutionsForTesting(); + const turn = await acquireTurnExecutionWithWait(null); + assert.equal(turn.acquired, true); +}); + +test("waits for the in-flight turn to release, then acquires", async () => { + clearTurnExecutionsForTesting(); + const first = acquireTurnExecution("k"); + assert.equal(first.acquired, true); + setTimeout(() => first.release(), 120); + const waits: string[] = []; + const second = await acquireTurnExecutionWithWait("k", { + timeoutMs: 2_000, + pollMs: 40, + log: { warn: (tag, msg) => waits.push(`${tag}:${msg}`) }, + traceId: "cid-1", + }); + assert.equal(second.acquired, true); + assert.ok(waits.some((w) => w.startsWith("TURN_GUARD:duplicate waiting cid=cid-1"))); +}); + +test("falls back to the blocked result after the timeout", async () => { + clearTurnExecutionsForTesting(); + const first = acquireTurnExecution("k"); + assert.equal(first.acquired, true); + const started = Date.now(); + const second = await acquireTurnExecutionWithWait("k", { timeoutMs: 150, pollMs: 40 }); + assert.equal(second.acquired, false); + assert.ok(Date.now() - started >= 140); + first.release(); +}); + +test("stops waiting when the caller aborts", async () => { + clearTurnExecutionsForTesting(); + const first = acquireTurnExecution("k"); + assert.equal(first.acquired, true); + const controller = new AbortController(); + setTimeout(() => controller.abort(), 80); + const second = await acquireTurnExecutionWithWait("k", { + timeoutMs: 10_000, + pollMs: 40, + signal: controller.signal, + }); + assert.equal(second.acquired, false); + first.release(); +});