Backport: fix(provider-utils): make ReadableStream.cancel() properly finalize async iterators (#9881)

This is an automated backport of #9834 to the release-v5.0 branch.

Co-authored-by: Dmitriy Osipov <75555523+teunlao@users.noreply.github.com>

Author: vercel-ai-sdk[bot]

.changeset/fix-async-iterator-cancel-cleanup.md

@@ -0,0 +1,5 @@
+---
+'@ai-sdk/provider-utils': patch
+---
+
+fix(provider-utils): make ReadableStream.cancel() properly finalize async iterators

packages/provider-utils/src/convert-async-iterator-to-readable-stream.test.ts

@@ -0,0 +1,78 @@
+import { convertAsyncIteratorToReadableStream } from './convert-async-iterator-to-readable-stream';
+import { describe, it, expect } from 'vitest';
+
+async function* makeGenerator(onFinally: () => void) {
+  try {
+    let i = 0;
+    while (true) {
+      await new Promise(r => setTimeout(r, 0));
+      yield i++;
+    }
+  } finally {
+    onFinally();
+  }
+}
+
+describe('convertAsyncIteratorToReadableStream', () => {
+  it('calls iterator.return() on cancel and triggers finally', async () => {
+    let finallyCalled = false;
+    const it = makeGenerator(() => {
+      finallyCalled = true;
+    });
+    const stream = convertAsyncIteratorToReadableStream(it);
+    const reader = stream.getReader();
+
+    await reader.read();
+
+    await reader.cancel('stop');
+
+    // give microtasks a tick for finally to run
+    await new Promise(r => setTimeout(r, 0));
+
+    expect(finallyCalled).toBe(true);
+  });
+
+  it('does not enqueue further values after cancel', async () => {
+    const it = makeGenerator(() => {});
+    const stream = convertAsyncIteratorToReadableStream(it);
+    const reader = stream.getReader();
+
+    await reader.read();
+    await reader.cancel('stop');
+
+    const { done, value } = await reader.read();
+    expect(done).toBe(true);
+    expect(value).toBeUndefined();
+  });
+
+  it('works with iterator without return() method', async () => {
+    const it: AsyncIterator<number> = {
+      async next() {
+        return { value: 42, done: false };
+      },
+    };
+    const stream = convertAsyncIteratorToReadableStream(it);
+    const reader = stream.getReader();
+
+    const { value } = await reader.read();
+    expect(value).toBe(42);
+
+    await expect(reader.cancel()).resolves.toBeUndefined();
+  });
+
+  it('ignores errors from iterator.return()', async () => {
+    const it: AsyncIterator<number> = {
+      async next() {
+        return { value: 1, done: false };
+      },
+      async return() {
+        throw new Error('return() failed');
+      },
+    };
+    const stream = convertAsyncIteratorToReadableStream(it);
+    const reader = stream.getReader();
+
+    await reader.read();
+    await expect(reader.cancel()).resolves.toBeUndefined();
+  });
+});

packages/provider-utils/src/convert-async-iterator-to-readable-stream.ts

@@ -8,6 +8,8 @@
 export function convertAsyncIteratorToReadableStream<T>(
   iterator: AsyncIterator<T>,
 ): ReadableStream<T> {
+  let cancelled = false;
+
   return new ReadableStream<T>({
     /**
      * Called when the consumer wants to pull more data from the stream.
@@ -16,6 +18,7 @@ export function convertAsyncIteratorToReadableStream<T>(
      * @returns {Promise<void>}
      */
     async pull(controller) {
+      if (cancelled) return;
       try {
         const { value, done } = await iterator.next();
         if (done) {
@@ -30,6 +33,15 @@ export function convertAsyncIteratorToReadableStream<T>(
     /**
      * Called when the consumer cancels the stream.
      */
-    cancel() {},
+    async cancel(reason?: unknown) {
+      cancelled = true;
+      if (iterator.return) {
+        try {
+          await iterator.return(reason);
+        } catch {
+          // intentionally ignore errors during cancellation
+        }
+      }
+    },
   });
 }