fix: swallow reader.cancel rejection when stream source already errored

.changeset/fix-stream-cancel-abort-error.md

@@ -0,0 +1,7 @@
+---
+"@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)
+
+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.

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<Uint8Array>;
+  const body = new ReadableStream<Uint8Array>({
+    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<typeof createOpenRouter>) =>
+        provider.chat('openai/gpt-4o-mini').doStream({ prompt: TEST_PROMPT }),
+    },
+    {
+      name: 'completion',
+      chunk: completionChunk,
+      start: (provider: ReturnType<typeof createOpenRouter>) =>
+        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<LanguageModelV4StreamPart> = 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([]);
+      });
+    }
+  }
+});

src/utils/with-stream-error-handling.test.ts

@@ -106,6 +106,65 @@ 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<string>;
+      const source = new ReadableStream<string>({
+        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 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<string>();
+    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);

src/utils/with-stream-error-handling.ts

@@ -3,23 +3,37 @@ export function withStreamErrorHandling<T>(
   onError: (error: unknown) => void,
 ): ReadableStream<T> {
   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<T>({
     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) {
-      reader.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.
+      return reader.cancel(reason).catch(() => {});
     },
   });
 }