Skip to content

fix(internal): cancel source stream when AsyncIterableStream stops early #1365

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
VihaanAgarwal wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(internal): cancel source stream when AsyncIterableStream stops early #1365

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions .changeset/async-iterable-stream-cancel.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
---
"@voltagent/internal": patch
---

Cancel the source stream when an `AsyncIterableStream` is iterated and stopped early

`createAsyncIterableStream` returned an async iterator with no `return` method, so breaking out of a `for await...of` loop early left the reader lock held and never cancelled the underlying stream. It now implements `return` to cancel the reader, matching the built-in `ReadableStream` async iterator and tearing down the upstream when iteration stops early.
22 changes: 22 additions & 0 deletions packages/internal/src/utils/async-iterable-stream.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,4 +31,26 @@ describe("createAsyncIterableStream()", () => {

expect(await convertReadableStreamToArray(asyncIterableStream)).toEqual(testData);
});

it("should cancel the source stream when async iteration stops early", async () => {
let cancelled = false;
const source = new ReadableStream<number>({
start(controller) {
controller.enqueue(1);
controller.enqueue(2);
controller.enqueue(3);
},
cancel() {
cancelled = true;
},
});

const asyncIterableStream = createAsyncIterableStream(source);

for await (const _chunk of asyncIterableStream) {
break; // stop after the first chunk
}

expect(cancelled).toBe(true);
});
});
7 changes: 7 additions & 0 deletions packages/internal/src/utils/async-iterable-stream.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,13 @@ export function createAsyncIterableStream<T>(source: ReadableStream<T>): AsyncIt
const { done, value } = await reader.read();
return done ? { done: true, value: undefined } : { done: false, value };
},
// Called when the consumer stops early (break/throw/return). Mirror the
// built-in ReadableStream async iterator and cancel the source so the
// upstream is torn down instead of leaking the reader lock.
async return(): Promise<IteratorResult<T>> {
await reader.cancel();
return { done: true, value: undefined };
Comment on lines +45 to +47

@coderabbitai coderabbitai Bot Jun 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🌐 Web query:

Does ReadableStreamDefaultReader.cancel() release the reader lock automatically, or must releaseLock() still be called to allow subsequent getReader() calls?

💡 Result:

No, ReadableStreamDefaultReader.cancel does not automatically release the reader lock [1][2]. The cancel method signals that you are no longer interested in the stream and causes the stream to be cancelled, but it leaves the reader active and the stream locked [1][2]. To allow subsequent getReader calls on the same stream, you must explicitly call releaseLock after calling cancel [1][3]. If you no longer need the reader after cancelling, the standard approach is to call both methods [1]: reader.cancel; reader.releaseLock; Calling releaseLock is the necessary step to make the stream available for a new reader [3]. If you attempt to acquire a new reader while the original reader still holds the lock, the stream will remain locked and the new getReader call will throw a TypeError [4][3].

Citations:

Release the reader lock after canceling.

reader.cancel() does not release the lock, so early exits can still leave the ReadableStreamDefaultReader attached and block later getReader() calls. Wrap this in finally and call reader.releaseLock().

Suggested fix 
      async return(): Promise<IteratorResult<T>> {
-        await reader.cancel();
-        return { done: true, value: undefined };
+        try {
+          await reader.cancel();
+        } finally {
+          reader.releaseLock();
+        }
+        return { done: true, value: undefined };
      },
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
async return(): Promise<IteratorResult<T>> {
await reader.cancel();
return { done: true, value: undefined };
async return(): Promise<IteratorResult<T>> {
try {
await reader.cancel();
} finally {
reader.releaseLock();
}
return { done: true, value: undefined };
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/internal/src/utils/async-iterable-stream.ts` around lines 45 - 47,
The async iterator’s return path in async-iterable-stream.ts cancels the stream
but leaves the ReadableStreamDefaultReader lock held, which can block later
getReader() calls. Update the return() implementation on the iterator to ensure
reader.releaseLock() is always called after reader.cancel(), ideally by wrapping
the cancel logic in a try/finally. Use the existing return() method and reader
symbol in this stream wrapper to locate the fix.

},
};
};

Expand Down
Loading