From 4789297c79aa913819433ddb27487333a5f6d93a Mon Sep 17 00:00:00 2001 From: Andy Date: Sun, 1 Nov 2026 18:36:18 +0800 Subject: [PATCH 1/2] fix(sse): wait out in-flight duplicate turns instead of failing them with 409 --- open-sse/handlers/chatCore.ts | 4 +- .../handlers/chatCore/turnExecutionGuard.ts | 51 +++++++++++++++ .../turn-execution-guard-wait.test.ts | 62 +++++++++++++++++++ 3 files changed, 115 insertions(+), 2 deletions(-) create mode 100644 tests/unit/chatcore/turn-execution-guard-wait.test.ts 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..8e0a12459fc7 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 = 30_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(); +}); From da9b010333c8236f2e3a5f6c040f5107aeac4313 Mon Sep 17 00:00:00 2001 From: Andy Date: Sun, 1 Nov 2026 18:38:28 +0800 Subject: [PATCH 2/2] fix(sse): raise turn wait bound to 60s to cover observed stuck-original lifetimes --- open-sse/handlers/chatCore/turnExecutionGuard.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/open-sse/handlers/chatCore/turnExecutionGuard.ts b/open-sse/handlers/chatCore/turnExecutionGuard.ts index 8e0a12459fc7..fe48c35b3726 100644 --- a/open-sse/handlers/chatCore/turnExecutionGuard.ts +++ b/open-sse/handlers/chatCore/turnExecutionGuard.ts @@ -75,7 +75,7 @@ export function getTurnExecutionSnapshot(key: string) { return { retryCount: active.retryCount, ageMs: Math.max(0, Date.now() - active.startedAt) }; } -export const TURN_WAIT_TIMEOUT_MS = 30_000; +export const TURN_WAIT_TIMEOUT_MS = 60_000; export const TURN_WAIT_POLL_MS = 500; export type TurnExecutionWaitOptions = {