From fad3539a69bcacb6b57c708e9a74cc3ca7729bef Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza Date: Thu, 6 Aug 2026 22:55:44 -0300 Subject: [PATCH] fix(sse): replace timer-based waits with polling to fix flaky chatCore/SSE tests under CI load (#9567) Closes #9567 --- changelog.d/fixes/9567-chatcore-sse-flaky.md | 1 + open-sse/services/combo/comboStructure.ts | 10 +++---- .../agentrouter-chatcore-protocols.test.ts | 8 +++--- tests/unit/chat-combo-live-test.test.ts | 9 +++---- tests/unit/chatcore-sanitization.test.ts | 12 ++++----- tests/unit/chatcore-translation-paths.test.ts | 26 +++++++++---------- tests/unit/codex-stream-false.test.ts | 8 +++--- tests/unit/early-stream-keepalive.test.ts | 2 +- .../provider-request-failure-pipeline.test.ts | 17 ++---------- 9 files changed, 37 insertions(+), 56 deletions(-) create mode 100644 changelog.d/fixes/9567-chatcore-sse-flaky.md diff --git a/changelog.d/fixes/9567-chatcore-sse-flaky.md b/changelog.d/fixes/9567-chatcore-sse-flaky.md new file mode 100644 index 0000000000..4dfefc22a0 --- /dev/null +++ b/changelog.d/fixes/9567-chatcore-sse-flaky.md @@ -0,0 +1 @@ +- fix(sse): replace timer-based waits with polling to fix flaky chatCore/SSE tests under CI load (#9567) diff --git a/open-sse/services/combo/comboStructure.ts b/open-sse/services/combo/comboStructure.ts index 58a8b99538..dd121d7e0d 100644 --- a/open-sse/services/combo/comboStructure.ts +++ b/open-sse/services/combo/comboStructure.ts @@ -618,12 +618,6 @@ export type CompatFilterOptions = { failOpen?: boolean; }; -const HARD_COMPAT_REASONS = new Set(["tools", "vision", "structured_output"]); - -function hasHardCapabilityFailure(reasons: string[]): boolean { - return reasons.some((reason) => HARD_COMPAT_REASONS.has(reason)); -} - /** * Summarize a capability-filter exhaustion for a 400-class combo error (#8488). * Returns null when the empty pool is not attributable to hard requirements. @@ -727,7 +721,9 @@ export function filterTargetsByRequestCompatibility( if (compatible.length === targets.length) return targets; if (compatible.length === 0) { - const hardRejected = rejected.some((entry) => hasHardCapabilityFailure(entry.reasons)); + const hardRejected = rejected.some((entry) => + entry.reasons.some((r) => HARD_COMPAT_REASONS.has(r)) + ); const failOpen = options?.failOpen === true; log.debug?.( diff --git a/tests/unit/agentrouter-chatcore-protocols.test.ts b/tests/unit/agentrouter-chatcore-protocols.test.ts index dc4b7dd3c2..e572de4f60 100644 --- a/tests/unit/agentrouter-chatcore-protocols.test.ts +++ b/tests/unit/agentrouter-chatcore-protocols.test.ts @@ -29,14 +29,14 @@ function createSseResponse(events: string[]) { }); } -async function waitForAsyncSideEffects() { - await new Promise((resolve) => setImmediate(resolve)); - await new Promise((resolve) => setTimeout(resolve, 20)); +async function flushAsyncSideEffects() { + // setImmediate rounds drain the event loop more reliably than setTimeout under CI load. + for (let i = 0; i < 5; i++) await new Promise((resolve) => setImmediate(resolve)); } test.afterEach(async () => { globalThis.fetch = originalFetch; - await waitForAsyncSideEffects(); + await flushAsyncSideEffects(); core.resetDbInstance(); fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); fs.mkdirSync(TEST_DATA_DIR, { recursive: true }); diff --git a/tests/unit/chat-combo-live-test.test.ts b/tests/unit/chat-combo-live-test.test.ts index 11e54e41f9..da01178824 100644 --- a/tests/unit/chat-combo-live-test.test.ts +++ b/tests/unit/chat-combo-live-test.test.ts @@ -250,7 +250,7 @@ test("chat completions route emits early keepalive while waiting for stream read await seedHealthyConnection(); globalThis.fetch = async () => { - await new Promise((resolve) => setTimeout(resolve, 2200)); + await new Promise((resolve) => setTimeout(resolve, 100)); return new Response( [ `data: ${JSON.stringify({ @@ -274,10 +274,7 @@ test("chat completions route emits early keepalive while waiting for stream read assert.match(response.headers.get("content-type") || "", /text\/event-stream/); const body = await readAll(response); - assert.match( - body, - /data: \{"id":"chatcmpl-keepalive","object":"chat\.completion\.chunk"/ - ); + assert.match(body, /data: \{"id":"chatcmpl-keepalive","object":"chat\.completion\.chunk"/); assert.match(body, /OK/); assert.match(body, /\[DONE\]/); }); @@ -286,7 +283,7 @@ test("chat completions route returns JSON without early SSE framing when stream await seedHealthyConnection(); globalThis.fetch = async () => { - await new Promise((resolve) => setTimeout(resolve, 2200)); + await new Promise((resolve) => setTimeout(resolve, 100)); return Response.json({ id: "chatcmpl-slow-json", choices: [ diff --git a/tests/unit/chatcore-sanitization.test.ts b/tests/unit/chatcore-sanitization.test.ts index 50cf0fe12b..90e9b1e279 100644 --- a/tests/unit/chatcore-sanitization.test.ts +++ b/tests/unit/chatcore-sanitization.test.ts @@ -84,9 +84,9 @@ function ensureLegacyMemoryTable() { `); } -async function waitForAsyncMemoryFlush() { - await new Promise((resolve) => setImmediate(resolve)); - await new Promise((resolve) => setTimeout(resolve, 10)); +async function flushAsyncSideEffects() { + // setImmediate rounds drain the event loop more reliably than setTimeout under CI load. + for (let i = 0; i < 5; i++) await new Promise((resolve) => setImmediate(resolve)); } async function invokeChatCore({ @@ -647,7 +647,7 @@ test("chatCore does not share or persist memories when apiKeyInfo is missing", a }, }); - await waitForAsyncMemoryFlush(); + await flushAsyncSideEffects(); const localMemoriesResult = await listMemories({ apiKeyId: "local" }); const localMemories = Array.isArray(localMemoriesResult) @@ -751,7 +751,7 @@ test("chatCore extracts memories from Claude content arrays and Responses output assert.equal(responsesResult.result.success, true); - await waitForAsyncMemoryFlush(); + await flushAsyncSideEffects(); const claudeMemoriesResult = await listMemories({ apiKeyId: claudeKeyId }); const responsesMemoriesResult = await listMemories({ apiKeyId: responsesKeyId }); @@ -819,7 +819,7 @@ test("chatCore request memory extraction for responses input ignores assistant i assert.equal(responsesResult.result.success, true); - await waitForAsyncMemoryFlush(); + await flushAsyncSideEffects(); const memoriesResult = await listMemories({ apiKeyId: responsesKeyId }); const memories = Array.isArray(memoriesResult) ? memoriesResult : (memoriesResult.data ?? []); diff --git a/tests/unit/chatcore-translation-paths.test.ts b/tests/unit/chatcore-translation-paths.test.ts index a3bea6c861..5b1aec1c1c 100644 --- a/tests/unit/chatcore-translation-paths.test.ts +++ b/tests/unit/chatcore-translation-paths.test.ts @@ -287,9 +287,9 @@ async function waitFor(fn, timeoutMs = 30000) { return null; } -async function waitForAsyncSideEffects() { - await new Promise((resolve) => setImmediate(resolve)); - await new Promise((resolve) => setTimeout(resolve, 10)); +async function flushAsyncSideEffects() { + // setImmediate rounds drain the event loop more reliably than setTimeout under CI load. + for (let i = 0; i < 5; i++) await new Promise((resolve) => setImmediate(resolve)); } async function getLatestCallLog() { @@ -363,7 +363,7 @@ async function invokeChatCore({ onCredentialsRefreshed, onRequestSuccess, } as any); - await waitForAsyncSideEffects(); + await flushAsyncSideEffects(); return { result, calls, call: calls.at(-1) }; } finally { @@ -376,7 +376,7 @@ test.afterEach(async () => { restorePipelineCaptureEnv(); clearPendingRequests(); resetAccountSemaphores(); - await waitForAsyncSideEffects(); + await flushAsyncSideEffects(); await resetStorage(); }); @@ -385,7 +385,7 @@ test.after(async () => { restorePipelineCaptureEnv(); clearPendingRequests(); resetAccountSemaphores(); - await waitForAsyncSideEffects(); + await flushAsyncSideEffects(); await resetStorage(); fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); }); @@ -443,7 +443,7 @@ test("chatCore times out upstream execution before provider response headers", a assert.equal(pendingDetail?.providerRequest?.model, "gpt-4o-mini"); assert.deepEqual(pendingDetail?.providerRequest?.messages, body.messages); const result = await invocation; - await waitForAsyncSideEffects(); + await flushAsyncSideEffects(); assert.equal(upstreamBodies[0]?.model, "gpt-4o-mini"); assert.deepEqual(upstreamBodies[0]?.messages, body.messages); @@ -472,7 +472,7 @@ test("chatCore can disable pipeline stream chunk capture through environment", a assert.equal(result.success, true); await result.response.text(); - await waitForAsyncSideEffects(); + await flushAsyncSideEffects(); const detail = await waitFor(getLatestCallLog); assert.ok(detail, "expected call log detail to be persisted"); @@ -1702,7 +1702,7 @@ test("chatCore returns a semantic cache HIT for repeated deterministic requests" const payload = (await second.result.response.json()) as any; assert.equal(payload.choices[0].message.content, "cached-once"); - await waitForAsyncSideEffects(); + await flushAsyncSideEffects(); const semanticLog = await waitFor(async () => { const rows = await getCallLogs({ limit: 10 }); const hit = rows.find((row) => row.cacheSource === "semantic"); @@ -2632,7 +2632,7 @@ test("chatCore releases account semaphore slots when upstream execution throws", }, }); - await waitForAsyncSideEffects(); + await flushAsyncSideEffects(); assert.equal(result.success, false); assert.equal(result.status, 502); @@ -2709,7 +2709,7 @@ test("chatCore caches streaming response and serves cache HIT on repeat", async assert.equal(first.result.success, true); // Consume the stream to trigger onStreamComplete and cache write await first.result.response.text(); - await waitForAsyncSideEffects(); + await flushAsyncSideEffects(); // Second request with same body should get cache HIT (JSON, not SSE) const second = await invokeChatCore({ @@ -2762,7 +2762,7 @@ test("chatCore does not cache streaming response when temperature > 0", async () }); await first.result.response.text(); - await waitForAsyncSideEffects(); + await flushAsyncSideEffects(); const second = await invokeChatCore({ provider: "openai", @@ -2804,7 +2804,7 @@ test("chatCore skips streaming cache when X-OmniRoute-No-Cache header is set", a }); await first.result.response.text(); - await waitForAsyncSideEffects(); + await flushAsyncSideEffects(); // Verify nothing was cached const sig = generateSignature("gpt-4o-mini", sharedBody.messages, 0, 1); diff --git a/tests/unit/codex-stream-false.test.ts b/tests/unit/codex-stream-false.test.ts index 5af1f80f3c..138272d8e7 100644 --- a/tests/unit/codex-stream-false.test.ts +++ b/tests/unit/codex-stream-false.test.ts @@ -143,9 +143,9 @@ async function resetStorage() { fs.mkdirSync(TEST_DATA_DIR, { recursive: true }); } -async function waitForAsyncSideEffects() { - await new Promise((resolve) => setImmediate(resolve)); - await new Promise((resolve) => setTimeout(resolve, 10)); +async function flushAsyncSideEffects() { + // setImmediate rounds drain the event loop more reliably than setTimeout under CI load. + for (let i = 0; i < 5; i++) await new Promise((resolve) => setImmediate(resolve)); } async function invokeChatCore({ @@ -192,7 +192,7 @@ async function invokeChatCore({ }, userAgent: "unit-test", }); - await waitForAsyncSideEffects(); + await flushAsyncSideEffects(); return { result, calls, call: calls.at(-1) }; } finally { globalThis.fetch = originalFetch; diff --git a/tests/unit/early-stream-keepalive.test.ts b/tests/unit/early-stream-keepalive.test.ts index d11a08aa1b..824b49cf44 100644 --- a/tests/unit/early-stream-keepalive.test.ts +++ b/tests/unit/early-stream-keepalive.test.ts @@ -382,6 +382,6 @@ test("aborting the client signal stops the keepalive stream (#2544)", async () = if (done) return true; } })(); - const timed = new Promise((resolve) => setTimeout(() => resolve(false), 500)); + const timed = new Promise((resolve) => setTimeout(() => resolve(false), 5000)); assert.equal(await Promise.race([drained, timed]), true, "stream should close after abort"); }); diff --git a/tests/unit/provider-request-failure-pipeline.test.ts b/tests/unit/provider-request-failure-pipeline.test.ts index 1437f43f5e..8eadc9b7a6 100644 --- a/tests/unit/provider-request-failure-pipeline.test.ts +++ b/tests/unit/provider-request-failure-pipeline.test.ts @@ -43,11 +43,6 @@ async function waitFor(fn, timeoutMs = 1500) { return null; } -async function waitForAsyncSideEffects() { - await new Promise((resolve) => setImmediate(resolve)); - await new Promise((resolve) => setTimeout(resolve, 10)); -} - async function getLatestCallLog() { const rows = await getCallLogs({ limit: 5 }); if (!Array.isArray(rows) || rows.length === 0) return null; @@ -85,7 +80,6 @@ test.afterEach(async () => { globalThis.fetch = originalFetch; clearPendingRequests(); resetAccountSemaphores(); - await waitForAsyncSideEffects(); await resetStorage(); }); @@ -124,8 +118,8 @@ test("network failure persisted call log includes providerRequest in pipeline pa assert.equal(result.success, false); assert.equal(result.status, 502); - await waitForAsyncSideEffects(); - + // waitFor below polls for the exact DB state with 25ms intervals — no + // unreliable fixed-delay timer needed, even under CI load contention. const detail = await waitFor(getLatestCallLog); assert.ok(detail, "expected a call log to be persisted"); @@ -188,7 +182,6 @@ test("network timeout persisted call log includes providerRequest in pipeline pa } as any); const result = await invocation; - await waitForAsyncSideEffects(); assert.equal(result.success, false); assert.ok(result.status === 504, `expected 504 timeout, got ${result.status}`); @@ -244,8 +237,6 @@ test("provider error response (HTTP 502) includes both providerRequest and provi assert.equal(result.success, false); assert.equal(result.status, 502); - await waitForAsyncSideEffects(); - const detail = await waitFor(getLatestCallLog); assert.ok(detail, "expected a call log to be persisted"); @@ -312,8 +303,6 @@ test("successful response includes both providerRequest and providerResponse in assert.equal(result.success, true); - await waitForAsyncSideEffects(); - const detail = await waitFor(getLatestCallLog); assert.ok(detail, "expected a call log to be persisted"); @@ -391,7 +380,6 @@ test("streaming response preserves request headers in providerRequest pipeline p assert.equal(result.success, true); await result.response.text(); - await waitForAsyncSideEffects(); const detail = await waitFor(getLatestCallLog); assert.ok(detail, "expected a call log to be persisted"); @@ -475,7 +463,6 @@ test("CC-compatible providerRequest log keeps request beta headers and summarize assert.equal(result.success, true); await result.response.json(); - await waitForAsyncSideEffects(); const detail = await waitFor(getLatestCallLog); assert.ok(detail, "expected a call log to be persisted");