diff --git a/open-sse/services/qoderCli.ts b/open-sse/services/qoderCli.ts index 507e92af8c..c2a9bd4dc6 100644 --- a/open-sse/services/qoderCli.ts +++ b/open-sse/services/qoderCli.ts @@ -407,6 +407,10 @@ const QODER_JOB_TOKEN_MIN_TTL_MS = 60 * 1000; type QoderJobTokenCacheEntry = { jobToken: string; expiresAt: number }; const qoderJobTokenCache = new Map(); +const qoderJobTokenPending = new Map< + string, + Promise<{ jobToken: string; expiresInMs: number } | null> +>(); type FetchLike = (input: string, init?: Record) => Promise; @@ -482,7 +486,14 @@ export async function resolveQoderJobToken( const cached = qoderJobTokenCache.get(trimmed); if (cached && cached.expiresAt > now) return cached.jobToken; - const exchanged = await exchangeQoderJobToken(trimmed, options); + let pending = qoderJobTokenPending.get(trimmed); + if (!pending) { + pending = exchangeQoderJobToken(trimmed, options).finally(() => { + qoderJobTokenPending.delete(trimmed); + }); + qoderJobTokenPending.set(trimmed, pending); + } + const exchanged = await pending; if (!exchanged) return trimmed; // graceful fallback — keep prior behavior qoderJobTokenCache.set(trimmed, { jobToken: exchanged.jobToken, @@ -494,6 +505,7 @@ export async function resolveQoderJobToken( /** Test-only: clear the job-token cache so unit tests don't leak state. */ export function __clearQoderJobTokenCache(): void { qoderJobTokenCache.clear(); + qoderJobTokenPending.clear(); } export async function validateQoderCliPat({ diff --git a/tests/unit/qoder-jobtoken-exchange-4683.test.ts b/tests/unit/qoder-jobtoken-exchange-4683.test.ts index e3a24e4d0c..b9eb6be5ec 100644 --- a/tests/unit/qoder-jobtoken-exchange-4683.test.ts +++ b/tests/unit/qoder-jobtoken-exchange-4683.test.ts @@ -76,6 +76,34 @@ test("#4683 resolveQoderJobToken exchanges a pt-* once and caches the jt-*", asy __clearQoderJobTokenCache(); }); +test("#4683 resolveQoderJobToken coalesces concurrent pt-* exchanges", async () => { + __clearQoderJobTokenCache(); + let fetchCount = 0; + let releaseExchange: (() => void) | undefined; + let markExchangeStarted: (() => void) | undefined; + const exchangeStarted = new Promise((resolveStarted) => { + markExchangeStarted = resolveStarted; + }); + const fetchImpl = async () => { + fetchCount += 1; + markExchangeStarted?.(); + await new Promise((release) => { + releaseExchange = release; + }); + return jsonResponse({ job_token: "jt-shared", expires_in: 86400 }); + }; + + const resolves = Array.from({ length: 8 }, () => + resolveQoderJobToken("pt-concurrent", { fetchImpl, now: 1_000 }) + ); + await exchangeStarted; + releaseExchange?.(); + const tokens = await Promise.all(resolves); + assert.deepEqual(tokens, Array(8).fill("jt-shared")); + assert.equal(fetchCount, 1, "concurrent resolves must share one upstream exchange"); + __clearQoderJobTokenCache(); +}); + test("#4683 resolveQoderJobToken passes a jt-* through without exchanging", async () => { __clearQoderJobTokenCache(); let fetchCount = 0;