@@ -34,7 +34,10 @@ const {
   isPromise,
   isUint8Array,
 } = require('internal/util/types');
-const { AbortController } = require('internal/abort_controller');
+const {
+  AbortController,
+  AbortSignal,
+} = require('internal/abort_controller');
 
 const {
   arrayBufferViewToUint8Array,
@@ -862,8 +865,33 @@ function pull(source, ...args) {
 
   return {
     __proto__: null,
-    async *[SymbolAsyncIterator]() {
-      yield* createAsyncPipeline(from(source), transforms, signal);
+    [SymbolAsyncIterator]() {
+      const controller = new AbortController();
+      const iteratorSignal = signal === undefined ?
+        controller.signal : AbortSignal.any([signal, controller.signal]);
+
+      async function* pipeline() {
+        yield* createAsyncPipeline(from(source), transforms, iteratorSignal);
+      }
+      const iterator = pipeline();
+
+      return {
+        __proto__: null,
+        next(value) {
+          return iterator.next(value);
+        },
+        return(value) {
+          controller.abort(lazyDOMException('Aborted', 'AbortError'));
+          return iterator.return(value);
+        },
+        throw(error) {
+          controller.abort(error);
+          return iterator.throw(error);
+        },
+        [SymbolAsyncIterator]() {
+          return this;
+        },
+      };
     },
   };
 }
 
@@ -170,6 +170,35 @@ async function testPullSignalAbortWhileSourceNextPending() {
   await assert.rejects(next, { name: 'AbortError' });
 }
 
+async function testPullReturnWhileSourceNextPending() {
+  let startNext;
+  const nextStarted = new Promise((resolve) => { startNext = resolve; });
+  const source = {
+    [Symbol.asyncIterator]() {
+      return {
+        next() {
+          startNext();
+          return new Promise(() => {});
+        },
+      };
+    },
+  };
+
+  const iter = pull(source)[Symbol.asyncIterator]();
+  const next = assert.rejects(iter.next(), { name: 'AbortError' });
+  await nextStarted;
+
+  const timeout = {};
+  const result = await Promise.race([
+    iter.return(),
+    new Promise((resolve) => setImmediate(resolve, timeout)),
+  ]);
+
+  assert.notStrictEqual(result, timeout);
+  assert.deepStrictEqual(result, { value: undefined, done: true });
+  await next;
+}
+
 async function testPullSignalAbortWithTransformWhileSourceNextPending() {
   const source = {
     [Symbol.asyncIterator]() {
@@ -417,6 +446,7 @@ async function testTransformOptionsNotShared() {
     testTapCallbackError(),
     testPullSignalAbortMidIteration(),
     testPullSignalAbortWhileSourceNextPending(),
+    testPullReturnWhileSourceNextPending(),
     testPullSignalAbortWithTransformWhileSourceNextPending(),
     testPullConsumerBreakCleanup(),
     testPullTransformReturnsPromise(),