Skip to content

Commit dc09b59

Browse files
easonliang28easonliang28
committed
Address pre-merge warnings: guard settleOnAbort's early return; share Qwen credential loads
settleOnAbort's already-aborted early return handed the caller the abort error without attaching a rejection handler to the pending promise, exposing a rejecting pending promise as an unhandled rejection. A no-op rejection guard now runs before the early return (same shape as rejectOnAbort's existing guard). New test aborts with a rejecting pending promise and asserts no unhandledRejection fires while the abort error surfaces (negative control: removing the guard fails exactly that test). QwenCodeHandler duplicated credential loading after cancellation: createMessage/completePrompt race ensureAuthenticated() with settleOnAbort, so a cancelled first request left its load running and a second request started a fresh credential read. ensureAuthenticated now shares the in-flight load promise the same way refreshAccessToken shares its refresh, cleared in a finally once the load settles, so a failed load is retried by the next request. New tests: concurrent cancel+retry reuses one credential read; success and failure both clear the shared promise (negative control: breaking the reuse guard fails exactly the reuse test; the test releases the hung read in a finally so a failing run cannot leak the pending load into later tests). Review test-fidelity threads: the two qwen-code-native-tools 401 refresh tests now abort only after the refresh wait is established (fetch never settles; drain until the fetch starts) instead of aborting synchronously inside the fetch mock, which exercised only the already-aborted fast path; the two qwen-code listener tests now assert the exact registered listener reference is removed instead of expect.any(Function). Verification at this head: 128/128 green across abort-signal, qwen-code, qwen-code-native-tools and lm-studio specs; tsc --noEmit 0 errors; eslint --max-warnings=0 clean with suppression counts unchanged; prettier clean.
1 parent 3b998a5 commit dc09b59

5 files changed

Lines changed: 163 additions & 13 deletions

File tree

‎src/api/providers/__tests__/qwen-code-native-tools.spec.ts‎

Lines changed: 30 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -770,17 +770,29 @@ describe("QwenCodeHandler Native Tools", () => {
770770

771771
it("should not retry after 401 when the abort signal fires during the refresh", async () => {
772772
const external = new AbortController()
773-
const fetchMock = vi.fn().mockImplementation(async () => {
774-
external.abort() // simulate Stop pressed while the token refresh is in flight
775-
return tokenResponse()
776-
})
773+
// The token endpoint never responds. The abort must fire only
774+
// after the refresh wait is established (settleOnAbort has
775+
// registered its listener): aborting inside the fetch mock would
776+
// land before the wait exists and only exercise the
777+
// already-aborted fast path, not the in-flight cut.
778+
const fetchMock = vi.fn().mockImplementation(() => new Promise<never>(() => {}))
777779
vi.stubGlobal("fetch", fetchMock)
778780
mockCreate.mockRejectedValueOnce(unauthorizedError())
779781

780782
const stream = handler.createMessage("test prompt", [], { taskId: "t1", abortSignal: external.signal })
783+
// Start consuming before draining so the 401 attempt fails and
784+
// the refresh fetch starts while microtasks are drained.
785+
const pending = collectStream(stream)
786+
for (let i = 0; i < 50 && fetchMock.mock.calls.length === 0; i++) {
787+
await Promise.resolve()
788+
}
789+
expect(fetchMock).toHaveBeenCalledTimes(1)
790+
791+
external.abort() // Stop pressed while the token refresh wait is established
792+
781793
let caught: unknown
782794
try {
783-
await collectStream(stream)
795+
await pending
784796
} catch (error) {
785797
caught = error
786798
}
@@ -922,16 +934,24 @@ describe("QwenCodeHandler Native Tools", () => {
922934

923935
it("should not retry after 401 when the abort signal fires during the refresh", async () => {
924936
const external = new AbortController()
925-
const fetchMock = vi.fn().mockImplementation(async () => {
926-
external.abort() // simulate Stop pressed while the token refresh is in flight
927-
return tokenResponse()
928-
})
937+
// Same established-wait pattern as the createMessage case: the
938+
// token endpoint never responds and the abort lands only after
939+
// settleOnAbort has registered its listener on the real wait.
940+
const fetchMock = vi.fn().mockImplementation(() => new Promise<never>(() => {}))
929941
vi.stubGlobal("fetch", fetchMock)
930942
mockCreate.mockRejectedValueOnce(unauthorizedError())
931943

944+
const promise = handler.completePrompt("hi", { abortSignal: external.signal })
945+
for (let i = 0; i < 50 && fetchMock.mock.calls.length === 0; i++) {
946+
await Promise.resolve()
947+
}
948+
expect(fetchMock).toHaveBeenCalledTimes(1)
949+
950+
external.abort() // Stop pressed while the token refresh wait is established
951+
932952
let caught: unknown
933953
try {
934-
await handler.completePrompt("hi", { abortSignal: external.signal })
954+
await promise
935955
} catch (error) {
936956
caught = error
937957
}

‎src/api/providers/__tests__/qwen-code.spec.ts‎

Lines changed: 85 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -284,10 +284,86 @@ describe("QwenCodeHandler abort wiring", () => {
284284
})
285285
})
286286

287+
describe("credential load sharing", () => {
288+
it("reuses the in-flight credential load when a cancelled request is followed by a retry", async () => {
289+
// The credential read hangs until released: the first request's Stop
290+
// cuts only its wait, the load keeps running, and a second request
291+
// that starts while the load is still in flight must reuse it
292+
// instead of issuing a fresh credential read.
293+
const creds = {
294+
access_token: "test-access-token",
295+
refresh_token: "test-refresh-token",
296+
token_type: "Bearer",
297+
expiry_date: Date.now() + 3600000,
298+
resource_url: "https://dashscope.aliyuncs.com/compatible-mode/v1",
299+
}
300+
let releaseLoad: ((value: string) => void) | undefined
301+
vi.mocked(fs.readFile).mockImplementation(
302+
() =>
303+
new Promise<string>((resolve) => {
304+
releaseLoad = resolve
305+
}),
306+
)
307+
// Release the hung read even when an assertion fails, so a failing
308+
// run cannot leak the pending load (and the queued create mock)
309+
// into later tests.
310+
try {
311+
const first = new AbortController()
312+
const firstRun = collectStream(
313+
handler.createMessage("test prompt", [], { taskId: "t1", abortSignal: first.signal }),
314+
).catch((error: unknown) => error)
315+
for (let i = 0; i < 20 && vi.mocked(fs.readFile).mock.calls.length === 0; i++) {
316+
await Promise.resolve()
317+
}
318+
expect(fs.readFile).toHaveBeenCalledTimes(1)
319+
320+
first.abort() // Stop lands while the credential load is in flight
321+
const firstError = await firstRun
322+
expect((firstError as Error).name).toBe("AbortError")
323+
324+
// A second request starting while the first load still runs reuses it.
325+
mockCreate.mockResolvedValueOnce(asyncStreamFrom([{ choices: [{ delta: { content: "ok" } }] }]))
326+
const secondRun = collectStream(handler.createMessage("test prompt", [], { taskId: "t2" })).catch(
327+
(error: unknown) => error,
328+
)
329+
await Promise.resolve() // let the second request enter the shared wait
330+
expect(fs.readFile).toHaveBeenCalledTimes(1) // no duplicate credential read
331+
332+
releaseLoad!(JSON.stringify(creds)) // the shared load finishes for both waiters
333+
const chunks = await secondRun
334+
expect(chunks).toContainEqual({ type: "text", text: "ok" })
335+
expect(fs.readFile).toHaveBeenCalledTimes(1)
336+
// The settled load is cleared so later failures can start a fresh read.
337+
expect(handler["credentialLoad"]).toBeNull()
338+
} finally {
339+
releaseLoad?.(JSON.stringify(creds))
340+
}
341+
})
342+
343+
it("clears the shared credential load after a failure so the next request retries the load", async () => {
344+
vi.mocked(fs.readFile).mockRejectedValueOnce(new Error("read failed"))
345+
let caught: unknown
346+
try {
347+
await handler.completePrompt("hi")
348+
} catch (error) {
349+
caught = error
350+
}
351+
expect((caught as Error).message).toMatch(/Failed to load Qwen OAuth credentials/)
352+
expect(handler["credentialLoad"]).toBeNull()
353+
354+
// The settled (failed) load must be cleared: the next request starts
355+
// a fresh read rather than reusing the rejected promise.
356+
mockCreate.mockResolvedValueOnce({ choices: [{ message: { content: "ok" } }] })
357+
expect(await handler.completePrompt("hi")).toBe("ok")
358+
expect(fs.readFile).toHaveBeenCalledTimes(2)
359+
})
360+
})
361+
287362
describe("createMessage", () => {
288363
it("passes a request-local signal and bridges the external abort", async () => {
289364
const external = new AbortController()
290365
const addSpy = vi.spyOn(external.signal, "addEventListener")
366+
const removeSpy = vi.spyOn(external.signal, "removeEventListener")
291367
mockCreate.mockResolvedValueOnce(asyncStreamFrom([{ choices: [{ delta: { content: "x" } }] }]))
292368

293369
const stream = handler.createMessage("test prompt", [], { taskId: "t1", abortSignal: external.signal })
@@ -297,12 +373,16 @@ describe("QwenCodeHandler abort wiring", () => {
297373
const opts = mockCreate.mock.calls[0][1]
298374
expect(opts?.signal).toBeInstanceOf(AbortSignal)
299375
expect(opts.signal).not.toBe(external.signal) // request-local, not the external signal
300-
expect(addSpy).toHaveBeenCalledWith("abort", expect.any(Function))
376+
const listener = addSpy.mock.calls.find(([type]) => type === "abort")?.[1]
377+
expect(typeof listener).toBe("function")
301378

302379
external.abort()
303380
expect(opts.signal.aborted).toBe(true) // the external abort is bridged to the SDK signal
304381

305382
await stream.next() // drain the generator
383+
// The exact registered listener must be removed - removing a
384+
// different function would leak the bridge on the shared Task signal.
385+
expect(removeSpy).toHaveBeenCalledWith("abort", listener)
306386
})
307387

308388
it("fast-fails with the abort contract error for a pre-aborted signal", async () => {
@@ -596,14 +676,17 @@ describe("QwenCodeHandler abort wiring", () => {
596676

597677
it("removes the external abort listener when the stream completes", async () => {
598678
const external = new AbortController()
679+
const addSpy = vi.spyOn(external.signal, "addEventListener")
599680
const removeSpy = vi.spyOn(external.signal, "removeEventListener")
600681
mockCreate.mockResolvedValueOnce(asyncStreamFrom([{ choices: [{ delta: { content: "ok" } }] }]))
601682

602683
await collectStream(
603684
handler.createMessage("test prompt", [], { taskId: "t1", abortSignal: external.signal }),
604685
)
605686

606-
expect(removeSpy).toHaveBeenCalledWith("abort", expect.any(Function))
687+
const listener = addSpy.mock.calls.find(([type]) => type === "abort")?.[1]
688+
expect(typeof listener).toBe("function")
689+
expect(removeSpy).toHaveBeenCalledWith("abort", listener)
607690
})
608691
})
609692

‎src/api/providers/qwen-code.ts‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@ export class QwenCodeHandler extends BaseProvider implements SingleCompletionHan
6464
private credentials: QwenOAuthCredentials | null = null
6565
private client: OpenAI | undefined
6666
private refreshPromise: Promise<QwenOAuthCredentials> | null = null
67+
private credentialLoad: Promise<QwenOAuthCredentials> | null = null
6768

6869
constructor(options: QwenCodeHandlerOptions) {
6970
super()
@@ -186,7 +187,18 @@ export class QwenCodeHandler extends BaseProvider implements SingleCompletionHan
186187

187188
private async ensureAuthenticated(): Promise<void> {
188189
if (!this.credentials) {
189-
this.credentials = await this.loadCachedQwenCredentials()
190+
// Share the in-flight credential load the same way refreshAccessToken
191+
// shares its refresh: a cancellation cuts only the caller's wait
192+
// (settleOnAbort), the load keeps running, and a second request that
193+
// starts while the first load continues reuses it instead of
194+
// duplicating credential I/O. Cleared once the load settles (success
195+
// or failure), so a failed load is retried by the next request.
196+
if (!this.credentialLoad) {
197+
this.credentialLoad = this.loadCachedQwenCredentials().finally(() => {
198+
this.credentialLoad = null
199+
})
200+
}
201+
this.credentials = await this.credentialLoad
190202
}
191203

192204
if (!this.isTokenValid(this.credentials)) {

‎src/api/providers/utils/__tests__/abort-signal.spec.ts‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -456,6 +456,37 @@ describe("settleOnAbort", () => {
456456
expect(await settleOnAbort(pending, controller.signal, "Test")).toBe(7)
457457
})
458458

459+
it("does not leak an unhandled rejection when the signal is already aborted and pending rejects", async () => {
460+
// The early return hands the caller the abort error and never awaits
461+
// pending; without a rejection guard on pending, its rejection would
462+
// surface as an unhandled rejection (Vitest fails the file on those).
463+
const rejections: unknown[] = []
464+
const onUnhandled = (reason: unknown): void => {
465+
rejections.push(reason)
466+
}
467+
process.on("unhandledRejection", onUnhandled)
468+
try {
469+
const controller = new AbortController()
470+
controller.abort()
471+
const failure = new Error("load failed")
472+
473+
let caught: unknown
474+
try {
475+
await settleOnAbort(Promise.reject(failure), controller.signal, "Test")
476+
} catch (error) {
477+
caught = error
478+
}
479+
await new Promise((resolve) => setTimeout(resolve, 10)) // let any unhandledRejection fire
480+
481+
expect(caught).toBeInstanceOf(Error)
482+
expect((caught as Error).name).toBe("AbortError")
483+
expect((caught as Error).message).toBe("The Test request was aborted")
484+
expect(rejections).toEqual([])
485+
} finally {
486+
process.off("unhandledRejection", onUnhandled)
487+
}
488+
})
489+
459490
it("rejects with the abort contract error when the signal aborts before the pending promise settles", async () => {
460491
const controller = new AbortController()
461492
let resolvePending!: (value: number) => void

‎src/api/providers/utils/abort-signal.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -184,6 +184,10 @@ export function settleOnAbort<T>(
184184
return pending
185185
}
186186
if (signal.aborted) {
187+
// Nobody awaits `pending` after this early return: attach a no-op
188+
// rejection handler so a rejecting pending promise cannot surface as
189+
// an unhandled rejection while its work runs to completion.
190+
void pending.catch(() => {})
187191
return Promise.reject(createAbortError(providerName))
188192
}
189193
return new Promise<T>((resolve, reject) => {

0 commit comments

Comments
 (0)