stream: reject pull on signal abort during flush

lib/internal/streams/iter/pull.js

@@ -764,6 +764,10 @@ async function* createAsyncPipeline(source, transforms, signal) {
       controller.signal.throwIfAborted();
       yield batch;
     }
+    // A transform can abort while completing without producing a final batch,
+    // for example when an async flush resolves to null. In that case the loop
+    // body has no opportunity to observe the abort.
+    controller.signal.throwIfAborted();
     completed = true;
   } catch (error) {
     if (!controller.signal.aborted) {

test/parallel/test-stream-iter-pull-async.js

@@ -295,6 +295,23 @@ async function testPullStatelessTransformFlushError() {
   }, { message: 'async flush boom' });
 }
 
+// An abort during an async flush must not be swallowed when the flush resolves
+// to null and therefore produces no final batch.
+async function testPullSignalAbortDuringAsyncFlush() {
+  const ac = new AbortController();
+  const reason = new Error('aborted during flush');
+  const transform = async (chunks) => {
+    if (chunks !== null) return chunks;
+    ac.abort(reason);
+    return null;
+  };
+
+  await assert.rejects(
+    () => text(pull(from('x'), transform, { signal: ac.signal })),
+    (error) => error === reason,
+  );
+}
+
 // Pull with a sync iterable source (not async)
 async function testPullWithSyncSource() {
   function* gen() {
@@ -409,6 +426,7 @@ async function testTransformOptionsNotShared() {
     testPullStatelessTransformFlush(),
     testPullConsecutiveStatelessTransformFlush(),
     testPullStatelessTransformFlushError(),
+    testPullSignalAbortDuringAsyncFlush(),
     testPullWithSyncSource(),
     testPullStringSource(),
     testTransformReturnsSingleUint8Array(),