From e2fb40a25614157c72161b8725af2c48685ffd2f Mon Sep 17 00:00:00 2001 From: Matt Apperson Date: Tue, 3 Nov 2026 15:30:38 -0500 Subject: [PATCH 1/2] fix: swallow reader.cancel rejection when stream source already errored Cancelling the wrapped stream after the source errored (e.g. aborted fetch) made reader.cancel() reject with no handler, crashing Node with an unhandled AbortError. Affects chat and completion streaming. Fixes #615 Fixes #625 --- .changeset/fix-stream-cancel-abort-error.md | 5 +++ src/utils/with-stream-error-handling.test.ts | 41 ++++++++++++++++++++ src/utils/with-stream-error-handling.ts | 5 ++- 3 files changed, 50 insertions(+), 1 deletion(-) create mode 100644 .changeset/fix-stream-cancel-abort-error.md diff --git a/.changeset/fix-stream-cancel-abort-error.md b/.changeset/fix-stream-cancel-abort-error.md new file mode 100644 index 00000000..1da87cac --- /dev/null +++ b/.changeset/fix-stream-cancel-abort-error.md @@ -0,0 +1,5 @@ +--- +"@openrouter/ai-sdk-provider": patch +--- + +Fix unhandled `AbortError` rejection when a streaming response is cancelled after its source has already errored (e.g. an aborted fetch) diff --git a/src/utils/with-stream-error-handling.test.ts b/src/utils/with-stream-error-handling.test.ts index ae606a73..fa73099e 100644 --- a/src/utils/with-stream-error-handling.test.ts +++ b/src/utils/with-stream-error-handling.test.ts @@ -106,6 +106,47 @@ describe('withStreamErrorHandling', () => { expect(cancelFn).toHaveBeenCalledWith('user abort'); }); + it('should not leak an unhandled rejection when cancelled after the source errored', async () => { + // Regression test for #615 / #625: an aborted fetch errors the response + // body while the consumer cancels the wrapper. Cancelling an errored + // reader rejects, and that rejection must not escape as unhandled. + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => { + unhandled.push(reason); + }; + process.on('unhandledRejection', onUnhandled); + + try { + let sourceController!: ReadableStreamDefaultController; + const source = new ReadableStream({ + start(controller) { + sourceController = controller; + }, + }); + const onError = vi.fn(); + const wrapped = withStreamErrorHandling(source, onError); + + // Fill the wrapper's queue so it has no pull in flight. + sourceController.enqueue('a'); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const abortError = new DOMException( + 'This operation was aborted', + 'AbortError', + ); + sourceController.error(abortError); + await expect(wrapped.cancel(abortError)).resolves.toBeUndefined(); + + // Give Node a chance to report any unhandled rejection. + await new Promise((resolve) => setTimeout(resolve, 10)); + + expect(unhandled).toEqual([]); + expect(onError).not.toHaveBeenCalled(); + } finally { + process.off('unhandledRejection', onUnhandled); + } + }); + it('should release the source reader lock on error', async () => { const error = new TypeError('terminated'); const source = createFailingStream(['a'], error); diff --git a/src/utils/with-stream-error-handling.ts b/src/utils/with-stream-error-handling.ts index 79056b61..be5233c6 100644 --- a/src/utils/with-stream-error-handling.ts +++ b/src/utils/with-stream-error-handling.ts @@ -19,7 +19,10 @@ export function withStreamErrorHandling( } }, cancel(reason) { - reader.cancel(reason); + // The source may already have errored (e.g. the fetch was aborted), in + // which case reader.cancel() rejects. Swallow it so it doesn't surface + // as an unhandled rejection. + return reader.cancel(reason).catch(() => {}); }, }); } From 2663716331dc4062ca804b20440eb28bee5e5037 Mon Sep 17 00:00:00 2001 From: Matt Apperson Date: Wed, 4 Nov 2026 09:35:27 -0500 Subject: [PATCH 2/2] test: add #615/#625 regression test and guard pull after cancel Add an offline e2e/issues regression test that drives chat and completion doStream through a mocked fetch, errors the body with an AbortError at queue depths 0-6, cancels, and asserts no unhandledRejection (fails at depth 2 without the fix). Also skip controller access in pull once the wrapper has been cancelled, so a read settling after cancel no longer passes a spurious 'Controller is already closed' TypeError to onError. --- .changeset/fix-stream-cancel-abort-error.md | 2 + ...-615-625-stream-cancel-after-abort.test.ts | 150 ++++++++++++++++++ src/utils/with-stream-error-handling.test.ts | 18 +++ src/utils/with-stream-error-handling.ts | 11 ++ 4 files changed, 181 insertions(+) create mode 100644 e2e/issues/issue-615-625-stream-cancel-after-abort.test.ts diff --git a/.changeset/fix-stream-cancel-abort-error.md b/.changeset/fix-stream-cancel-abort-error.md index 1da87cac..f7739c28 100644 --- a/.changeset/fix-stream-cancel-abort-error.md +++ b/.changeset/fix-stream-cancel-abort-error.md @@ -3,3 +3,5 @@ --- Fix unhandled `AbortError` rejection when a streaming response is cancelled after its source has already errored (e.g. an aborted fetch) + +Also stop reporting a spurious `Controller is already closed` error through the stream error handler when a stream is cancelled while a read is in flight. diff --git a/e2e/issues/issue-615-625-stream-cancel-after-abort.test.ts b/e2e/issues/issue-615-625-stream-cancel-after-abort.test.ts new file mode 100644 index 00000000..d9000083 --- /dev/null +++ b/e2e/issues/issue-615-625-stream-cancel-after-abort.test.ts @@ -0,0 +1,150 @@ +/** + * Regression test for GitHub issues #615 and #625 + * https://github.com/OpenRouterTeam/ai-sdk-provider/issues/615 + * https://github.com/OpenRouterTeam/ai-sdk-provider/issues/625 + * + * Reported error: Unhandled AbortError when a cancelled stream has already + * errored ("Stopping a chat can crash the process with an unhandled + * `AbortError`"). + * Affected: 3.1.0, doStream on chat models + * + * This test verifies that cancelling a doStream() result after the HTTP + * response body has errored with an AbortError does not produce an + * unhandledRejection, for both chat and completion models and for a range of + * buffered-but-unread SSE chunk depths. Fully offline: a custom `fetch` + * returns a Response whose body is controlled by the test. + */ +import type { + LanguageModelV4Prompt, + LanguageModelV4StreamPart, +} from '@ai-sdk/provider'; + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { createOpenRouter } from '@/src'; + +vi.mock('@/src/version', () => ({ + VERSION: '0.0.0-test', +})); + +const TEST_PROMPT: LanguageModelV4Prompt = [ + { role: 'user', content: [{ type: 'text', text: 'Hello' }] }, +]; + +const encoder = new TextEncoder(); + +function chatChunk(i: number): string { + return `data: ${JSON.stringify({ + id: 'test-1', + object: 'chat.completion.chunk', + created: 1234567890, + model: 'openai/gpt-4o-mini', + choices: [ + { index: 0, delta: { content: `token-${i}` }, finish_reason: null }, + ], + })}\n\n`; +} + +function completionChunk(i: number): string { + return `data: ${JSON.stringify({ + id: 'test-1', + object: 'text_completion', + created: 1234567890, + model: 'openai/gpt-4o-mini', + choices: [{ index: 0, text: `token-${i}`, finish_reason: null }], + })}\n\n`; +} + +function createControlledFetch() { + let bodyController!: ReadableStreamDefaultController; + const body = new ReadableStream({ + start(controller) { + bodyController = controller; + }, + }); + const fetch = vi.fn( + async () => + new Response(body, { + status: 200, + headers: { 'content-type': 'text/event-stream' }, + }), + ); + return { + fetch: fetch as unknown as typeof globalThis.fetch, + get body() { + return bodyController; + }, + }; +} + +const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); + +describe('Issue #615/#625: cancelling a stream whose source already errored', () => { + let unhandled: unknown[]; + const onUnhandled = (reason: unknown) => { + unhandled.push(reason); + }; + + beforeEach(() => { + unhandled = []; + process.on('unhandledRejection', onUnhandled); + }); + + afterEach(() => { + process.off('unhandledRejection', onUnhandled); + }); + + const cases = [ + { + name: 'chat', + chunk: chatChunk, + start: (provider: ReturnType) => + provider.chat('openai/gpt-4o-mini').doStream({ prompt: TEST_PROMPT }), + }, + { + name: 'completion', + chunk: completionChunk, + start: (provider: ReturnType) => + provider + .completion('openai/gpt-4o-mini') + .doStream({ prompt: TEST_PROMPT }), + }, + ] as const; + + for (const { name, chunk, start } of cases) { + for (let queued = 0; queued <= 6; queued++) { + it(`${name}: no unhandledRejection with ${queued} unread chunk(s) queued`, async () => { + const controlled = createControlledFetch(); + const provider = createOpenRouter({ + apiKey: 'test-api-key', + fetch: controlled.fetch, + }); + + const { stream } = await start(provider); + const typedStream: ReadableStream = stream; + + for (let i = 0; i < queued; i++) { + controlled.body.enqueue(encoder.encode(chunk(i))); + } + // Let the pipeline pull the queued chunks into its internal buffers. + await tick(); + await tick(); + + const abortError = new DOMException( + 'This operation was aborted', + 'AbortError', + ); + controlled.body.error(abortError); + // Let the error propagate through the pipe chain so the source the + // provider wraps is already errored when the consumer cancels. + await tick(); + await tick(); + await typedStream.cancel(abortError); + + // Give Node a chance to report any unhandled rejection. + await new Promise((resolve) => setTimeout(resolve, 20)); + + expect(unhandled).toEqual([]); + }); + } + } +}); diff --git a/src/utils/with-stream-error-handling.test.ts b/src/utils/with-stream-error-handling.test.ts index fa73099e..a272f013 100644 --- a/src/utils/with-stream-error-handling.test.ts +++ b/src/utils/with-stream-error-handling.test.ts @@ -147,6 +147,24 @@ describe('withStreamErrorHandling', () => { } }); + it('should not call onError when cancelled while a read is in flight', async () => { + // A source that never produces, so the wrapper's pull stays parked on + // reader.read() until the consumer cancels. + const source = new ReadableStream(); + const onError = vi.fn(); + const wrapped = withStreamErrorHandling(source, onError); + + const reader = wrapped.getReader(); + const pending = reader.read(); + await new Promise((resolve) => setTimeout(resolve, 0)); + + await reader.cancel('user abort'); + await expect(pending).resolves.toEqual({ done: true, value: undefined }); + await new Promise((resolve) => setTimeout(resolve, 10)); + + expect(onError).not.toHaveBeenCalled(); + }); + it('should release the source reader lock on error', async () => { const error = new TypeError('terminated'); const source = createFailingStream(['a'], error); diff --git a/src/utils/with-stream-error-handling.ts b/src/utils/with-stream-error-handling.ts index be5233c6..914a421e 100644 --- a/src/utils/with-stream-error-handling.ts +++ b/src/utils/with-stream-error-handling.ts @@ -3,22 +3,33 @@ export function withStreamErrorHandling( onError: (error: unknown) => void, ): ReadableStream { const reader = source.getReader(); + // Set once the consumer cancels. A read that was in flight at that point + // settles afterwards, and the controller is already closed, so pull must + // not touch it (or report the resulting TypeError via onError). + let cancelled = false; return new ReadableStream({ async pull(controller) { try { const { done, value } = await reader.read(); + if (cancelled) { + return; + } if (done) { controller.close(); } else { controller.enqueue(value); } } catch (err) { + if (cancelled) { + return; + } onError(err); reader.cancel().catch(() => {}); controller.close(); } }, cancel(reason) { + cancelled = true; // The source may already have errored (e.g. the fetch was aborted), in // which case reader.cancel() rejects. Swallow it so it doesn't surface // as an unhandled rejection.