From 1fe42f08851ee9f2edfe8fe513e9382eb7002b66 Mon Sep 17 00:00:00 2001 From: Christiaan Arnoldus Date: Tue, 28 Jul 2026 11:56:30 +0200 Subject: [PATCH 1/4] fix(ai-gateway): continue usage processing after stream errors --- .../src/lib/ai-gateway/processUsage.shared.ts | 20 +++++----- .../src/lib/ai-gateway/processUsage.test.ts | 37 ++++++++++++++++--- 2 files changed, 43 insertions(+), 14 deletions(-) diff --git a/apps/web/src/lib/ai-gateway/processUsage.shared.ts b/apps/web/src/lib/ai-gateway/processUsage.shared.ts index e33c5180e8..7bb009b6ae 100644 --- a/apps/web/src/lib/ai-gateway/processUsage.shared.ts +++ b/apps/web/src/lib/ai-gateway/processUsage.shared.ts @@ -91,8 +91,8 @@ export function isResponseInterruptedError(error: unknown): boolean { /** * Drains a ReadableStream of binary chunks, calling `onTextChunk` for each - * decoded piece of text. Handles client aborts and upstream timeouts gracefully - * and always releases the reader lock and ends `streamProcessingSpan`. + * decoded piece of text. Handles response body read failures gracefully and + * always releases the reader lock and ends `streamProcessingSpan`. * * Returns `true` if the stream was aborted before completion. */ @@ -106,16 +106,18 @@ export async function drainSseStream( let wasAborted = false; try { while (true) { - const { done, value } = await reader.read(); + let readResult: ReadableStreamReadResult; + try { + readResult = await reader.read(); + } catch { + wasAborted = true; + break; + } + + const { done, value } = readResult; if (done) break; onTextChunk(decoder.decode(value, { stream: true })); } - } catch (error) { - if (isResponseInterruptedError(error)) { - wasAborted = true; - } else { - throw error; - } } finally { reader.releaseLock(); streamProcessingSpan.end(); diff --git a/apps/web/src/lib/ai-gateway/processUsage.test.ts b/apps/web/src/lib/ai-gateway/processUsage.test.ts index 3710633f89..078ba31dd2 100644 --- a/apps/web/src/lib/ai-gateway/processUsage.test.ts +++ b/apps/web/src/lib/ai-gateway/processUsage.test.ts @@ -175,15 +175,29 @@ describe('parseMicrodollarUsageFromStream approval tests', () => { await verifyApproval(resultString, approvalFilePath); }); - test.each(['ResponseAborted', 'TimeoutError'])( + const interruptedStreamErrors: [name: string, error: Error][] = [ + [ + 'ResponseAborted', + Object.assign(new Error('Response interrupted'), { name: 'ResponseAborted' }), + ], + ['TimeoutError', Object.assign(new Error('Response interrupted'), { name: 'TimeoutError' })], + [ + 'Undici body timeout', + new TypeError('terminated', { + cause: Object.assign(new Error('Body Timeout Error'), { + name: 'BodyTimeoutError', + code: 'UND_ERR_BODY_TIMEOUT', + }), + }), + ], + ]; + + test.each(interruptedStreamErrors)( 'handles %s gracefully and returns partial data', - async errorName => { + async (_name, streamError) => { // Create a stream that emits some SSE data then fails. const partialSSEData = `data: {"id":"gen-123","model":"anthropic/claude-3-5-sonnet","choices":[{"delta":{"content":"Hello"}}]}\n\ndata: {"id":"gen-123","model":"anthropic/claude-3-5-sonnet","choices":[{"delta":{"content":" world"}}]}\n\n`; - const streamError = new Error('Response interrupted'); - streamError.name = errorName; - let pullCount = 0; const stream = new ReadableStream({ pull(controller) { @@ -211,6 +225,19 @@ describe('parseMicrodollarUsageFromStream approval tests', () => { } ); + test('does not swallow SSE parser errors', async () => { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: not-json\n\n')); + controller.close(); + }, + }); + + await expect( + parseMicrodollarUsageFromStream(stream, 'fake-user-id', undefined, 'openrouter', 200) + ).rejects.toThrow(SyntaxError); + }); + test.each([ ['chat_completions', 'ResponseAborted'], ['chat_completions', 'TimeoutError'], From 0f283748d596bb8197355486eeadbeb51ec78993 Mon Sep 17 00:00:00 2001 From: Christiaan Arnoldus Date: Tue, 28 Jul 2026 12:24:10 +0200 Subject: [PATCH 2/4] fix(ai-gateway): report unexpected stream read errors --- .../src/lib/ai-gateway/processUsage.shared.ts | 22 +++++++++++++++---- .../src/lib/ai-gateway/processUsage.test.ts | 1 + 2 files changed, 19 insertions(+), 4 deletions(-) diff --git a/apps/web/src/lib/ai-gateway/processUsage.shared.ts b/apps/web/src/lib/ai-gateway/processUsage.shared.ts index 7bb009b6ae..494ae98758 100644 --- a/apps/web/src/lib/ai-gateway/processUsage.shared.ts +++ b/apps/web/src/lib/ai-gateway/processUsage.shared.ts @@ -1,4 +1,4 @@ -import { captureMessage } from '@sentry/nextjs'; +import { captureException, captureMessage } from '@sentry/nextjs'; import type { Span } from '@sentry/nextjs'; import { toMicrodollars } from '../utils'; import { OPENROUTER_BYOK_COST_MULTIPLIER } from '@/lib/ai-gateway/processUsage.constants'; @@ -85,8 +85,19 @@ export function extractVercelIsByok( } export function isResponseInterruptedError(error: unknown): boolean { - if (typeof error !== 'object' || error === null || !('name' in error)) return false; - return error.name === 'ResponseAborted' || error.name === 'TimeoutError'; + if (typeof error !== 'object' || error === null) return false; + if ('name' in error && (error.name === 'ResponseAborted' || error.name === 'TimeoutError')) { + return true; + } + + if (!('cause' in error)) return false; + const cause = error.cause; + return ( + typeof cause === 'object' && + cause !== null && + 'code' in cause && + cause.code === 'UND_ERR_BODY_TIMEOUT' + ); } /** @@ -109,7 +120,10 @@ export async function drainSseStream( let readResult: ReadableStreamReadResult; try { readResult = await reader.read(); - } catch { + } catch (error) { + if (!isResponseInterruptedError(error)) { + captureException(error, { tags: { source: 'usage_stream_read' } }); + } wasAborted = true; break; } diff --git a/apps/web/src/lib/ai-gateway/processUsage.test.ts b/apps/web/src/lib/ai-gateway/processUsage.test.ts index 078ba31dd2..30fd5f946a 100644 --- a/apps/web/src/lib/ai-gateway/processUsage.test.ts +++ b/apps/web/src/lib/ai-gateway/processUsage.test.ts @@ -190,6 +190,7 @@ describe('parseMicrodollarUsageFromStream approval tests', () => { }), }), ], + ['unexpected read failure', new Error('Unexpected stream failure')], ]; test.each(interruptedStreamErrors)( From ea2fdc7f433f3cedeb65f92ca3ae49f2c52335c1 Mon Sep 17 00:00:00 2001 From: Christiaan Arnoldus Date: Tue, 28 Jul 2026 12:43:45 +0200 Subject: [PATCH 3/4] fix(ai-gateway): log all usage stream aborts --- .../src/lib/ai-gateway/processUsage.shared.ts | 33 ++++--------------- .../src/lib/ai-gateway/processUsage.test.ts | 14 +++++--- 2 files changed, 17 insertions(+), 30 deletions(-) diff --git a/apps/web/src/lib/ai-gateway/processUsage.shared.ts b/apps/web/src/lib/ai-gateway/processUsage.shared.ts index 494ae98758..b95d41cced 100644 --- a/apps/web/src/lib/ai-gateway/processUsage.shared.ts +++ b/apps/web/src/lib/ai-gateway/processUsage.shared.ts @@ -85,24 +85,13 @@ export function extractVercelIsByok( } export function isResponseInterruptedError(error: unknown): boolean { - if (typeof error !== 'object' || error === null) return false; - if ('name' in error && (error.name === 'ResponseAborted' || error.name === 'TimeoutError')) { - return true; - } - - if (!('cause' in error)) return false; - const cause = error.cause; - return ( - typeof cause === 'object' && - cause !== null && - 'code' in cause && - cause.code === 'UND_ERR_BODY_TIMEOUT' - ); + if (typeof error !== 'object' || error === null || !('name' in error)) return false; + return error.name === 'ResponseAborted' || error.name === 'TimeoutError'; } /** * Drains a ReadableStream of binary chunks, calling `onTextChunk` for each - * decoded piece of text. Handles response body read failures gracefully and + * decoded piece of text. Handles stream processing failures gracefully and * always releases the reader lock and ends `streamProcessingSpan`. * * Returns `true` if the stream was aborted before completion. @@ -117,21 +106,13 @@ export async function drainSseStream( let wasAborted = false; try { while (true) { - let readResult: ReadableStreamReadResult; - try { - readResult = await reader.read(); - } catch (error) { - if (!isResponseInterruptedError(error)) { - captureException(error, { tags: { source: 'usage_stream_read' } }); - } - wasAborted = true; - break; - } - - const { done, value } = readResult; + const { done, value } = await reader.read(); if (done) break; onTextChunk(decoder.decode(value, { stream: true })); } + } catch (error) { + captureException(error, { tags: { source: 'usage_stream_processing' } }); + wasAborted = true; } finally { reader.releaseLock(); streamProcessingSpan.end(); diff --git a/apps/web/src/lib/ai-gateway/processUsage.test.ts b/apps/web/src/lib/ai-gateway/processUsage.test.ts index 30fd5f946a..6e3a203dd2 100644 --- a/apps/web/src/lib/ai-gateway/processUsage.test.ts +++ b/apps/web/src/lib/ai-gateway/processUsage.test.ts @@ -226,7 +226,7 @@ describe('parseMicrodollarUsageFromStream approval tests', () => { } ); - test('does not swallow SSE parser errors', async () => { + test('handles SSE parser errors as aborted streams', async () => { const stream = new ReadableStream({ start(controller) { controller.enqueue(new TextEncoder().encode('data: not-json\n\n')); @@ -234,9 +234,15 @@ describe('parseMicrodollarUsageFromStream approval tests', () => { }, }); - await expect( - parseMicrodollarUsageFromStream(stream, 'fake-user-id', undefined, 'openrouter', 200) - ).rejects.toThrow(SyntaxError); + const result = await parseMicrodollarUsageFromStream( + stream, + 'fake-user-id', + undefined, + 'openrouter', + 200 + ); + + expect(result.hasError).toBe(true); }); test.each([ From 601df4027aafcc97dff4c2d1acada69c3700f75b Mon Sep 17 00:00:00 2001 From: Christiaan Arnoldus Date: Tue, 28 Jul 2026 12:53:20 +0200 Subject: [PATCH 4/4] fix(ai-gateway): avoid reporting usage stream aborts --- apps/web/src/lib/ai-gateway/processUsage.shared.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/apps/web/src/lib/ai-gateway/processUsage.shared.ts b/apps/web/src/lib/ai-gateway/processUsage.shared.ts index b95d41cced..f21734b5e4 100644 --- a/apps/web/src/lib/ai-gateway/processUsage.shared.ts +++ b/apps/web/src/lib/ai-gateway/processUsage.shared.ts @@ -1,6 +1,7 @@ -import { captureException, captureMessage } from '@sentry/nextjs'; +import { captureMessage } from '@sentry/nextjs'; import type { Span } from '@sentry/nextjs'; import { toMicrodollars } from '../utils'; +import { errorExceptInTest } from '@/lib/utils.server'; import { OPENROUTER_BYOK_COST_MULTIPLIER } from '@/lib/ai-gateway/processUsage.constants'; import type { NotYetCostedUsageStats, @@ -111,7 +112,7 @@ export async function drainSseStream( onTextChunk(decoder.decode(value, { stream: true })); } } catch (error) { - captureException(error, { tags: { source: 'usage_stream_processing' } }); + errorExceptInTest('[processUsage] treating stream processing error as aborted', error); wasAborted = true; } finally { reader.releaseLock();