@@ -186,8 +186,10 @@ async function normalizeBatch(raw) {
 // (Uint8Array subclass) and are yielded directly.
 const nop = () => {};
 
-async function* createBatchedAsyncIterator(stream, normalize) {
+function createBatchedAsyncIterator(stream, normalize) {
   let callback = nop;
+  let canceled = false;
+  let started = false;
 
   function next(resolve) {
     if (this === stream) {
@@ -198,54 +200,91 @@ async function* createBatchedAsyncIterator(stream, normalize) {
     }
   }
 
-  stream.on('readable', next);
-
-  let error;
-  const cleanup = eos(stream, { writable: false }, (err) => {
-    error = err ? aggregateTwoErrors(error, err) : null;
+  function wake() {
     callback();
     callback = nop;
-  });
+  }
 
-  try {
-    while (true) {
-      const chunk = stream.destroyed ? null : stream.read();
-      if (chunk !== null) {
-        const batch = [chunk];
-        while (batch.length < MAX_DRAIN_BATCH &&
-               stream._readableState?.length > 0) {
-          const c = stream.read();
-          if (c === null) break;
-          ArrayPrototypePush(batch, c);
-        }
-        if (normalize !== null) {
-          const result = await normalize(batch);
-          if (result !== null) {
-            yield result;
+  async function* generate() {
+    stream.on('readable', next);
+
+    let error;
+    const cleanup = eos(stream, { writable: false }, (err) => {
+      error = err ? aggregateTwoErrors(error, err) : null;
+      stream.removeListener('readable', next);
+      wake();
+    });
+    const cleanupAll = () => {
+      stream.removeListener('close', cleanupAll);
+      cleanup();
+    };
+    stream.on('close', cleanupAll);
+
+    try {
+      while (!canceled) {
+        const chunk = stream.destroyed ? null : stream.read();
+        if (chunk !== null) {
+          const batch = [chunk];
+          while (batch.length < MAX_DRAIN_BATCH &&
+                 stream._readableState?.length > 0) {
+            const c = stream.read();
+            if (c === null) break;
+            ArrayPrototypePush(batch, c);
+          }
+          if (normalize !== null) {
+            const result = await normalize(batch);
+            if (result !== null) {
+              yield result;
+            }
+          } else {
+            yield batch;
           }
+        } else if (error) {
+          throw error;
+        } else if (error === null) {
+          return;
         } else {
-          yield batch;
+          await new Promise(next);
         }
-      } else if (error) {
-        throw error;
-      } else if (error === null) {
-        return;
-      } else {
-        await new Promise(next);
       }
-    }
-  } catch (err) {
-    error = aggregateTwoErrors(error, err);
-    throw error;
-  } finally {
-    if (error === undefined ||
-        (stream._readableState?.autoDestroy)) {
-      destroyImpl.destroyer(stream, null);
-    } else {
+    } catch (err) {
+      error = aggregateTwoErrors(error, err);
+      throw error;
+    } finally {
       stream.removeListener('readable', next);
-      cleanup();
+      if (error === undefined ||
+          (stream._readableState?.autoDestroy)) {
+        destroyImpl.destroyer(stream, null);
+        if (stream._readableState?.closeEmitted) cleanupAll();
+      } else {
+        cleanupAll();
+      }
     }
   }
+
+  const iterator = generate();
+  const iteratorNext = iterator.next;
+  const iteratorReturn = iterator.return;
+  const iteratorThrow = iterator.throw;
+
+  iterator.next = function(value) {
+    started = true;
+    return FunctionPrototypeCall(iteratorNext, iterator, value);
+  };
+  iterator.return = function(value) {
+    canceled = true;
+    wake();
+    if (!started) stream.destroy();
+    return FunctionPrototypeCall(iteratorReturn, iterator, value);
+  };
+  iterator.throw = function(reason) {
+    canceled = true;
+    wake();
+    if (!started) stream.destroy();
+    return FunctionPrototypeCall(iteratorThrow, iterator, reason);
+  };
+
+  return iterator;
 }
 
 /**
 
@@ -7,6 +7,7 @@
 const common = require('../common');
 const assert = require('assert');
 const { Readable } = require('stream');
+const { setImmediate } = require('timers/promises');
 const {
   from,
   pull,
@@ -588,6 +589,38 @@ async function testAbortSignal() {
   assert.ok(chunks.length >= 2);
 }
 
+async function testReturnWhileReadIsPending() {
+  const readable = new Readable({ read() {} });
+  const iterator = from(readable)[Symbol.asyncIterator]();
+  const pending = iterator.next();
+  await setImmediate();
+
+  assert.deepStrictEqual(await iterator.return('stopped'), {
+    value: 'stopped',
+    done: true,
+  });
+  assert.deepStrictEqual(await pending, { value: undefined, done: true });
+  await setImmediate();
+  assert.strictEqual(readable.destroyed, true);
+  assert.strictEqual(readable.listenerCount('readable'), 0);
+}
+
+async function testAbortWhileReadIsPending() {
+  const readable = new Readable({ read() {} });
+  const controller = new AbortController();
+  const iterator = pull(readable, {
+    signal: controller.signal,
+  })[Symbol.asyncIterator]();
+  const pending = iterator.next();
+  await setImmediate();
+
+  controller.abort();
+  await assert.rejects(pending, { name: 'AbortError' });
+  await setImmediate();
+  assert.strictEqual(readable.destroyed, true);
+  assert.strictEqual(readable.listenerCount('readable'), 0);
+}
+
 // =============================================================================
 // kValidatedSource identity - from() returns same object for validated sources
 // =============================================================================
@@ -637,4 +670,6 @@ Promise.all([
   testDuplexStream(),
   testSetEncodingDynamic(),
   testAbortSignal(),
+  testReturnWhileReadIsPending(),
+  testAbortWhileReadIsPending(),
 ]).then(common.mustCall());
@@ -157,6 +157,30 @@ async function testFalsyThenableCleanupError() {
   assert.strictEqual(result.reason, reason);
 }
 
+async function testSynchronousIteratorReturn() {
+  let returnCalled = false;
+  const source = {
+    __proto__: null,
+    [Symbol.asyncIterator]() {
+      return {
+        __proto__: null,
+        next() { return kNeverResolves; },
+        return() {
+          returnCalled = true;
+          return { __proto__: null, done: true };
+        },
+      };
+    },
+  };
+  const readable = toReadable(source);
+  const { promise, resolve } = Promise.withResolvers();
+  readable.once('close', resolve);
+
+  readable.destroy();
+  await promise;
+  assert.strictEqual(returnCalled, true);
+}
+
 async function testFalsyCleanupGetterErrors() {
   for (const [symbol, create] of [
     [Symbol.asyncIterator, toReadable],
@@ -722,6 +746,7 @@ Promise.all([
   testErrorAsync(),
   testFalsyErrorAsync(),
   testFalsyThenableCleanupError(),
+  testSynchronousIteratorReturn(),
   testFalsyCleanupGetterErrors(),
   testEmptyAsync(),
   testEmptyBatchAsync(),