stream: destroy half-open sockets after iteration

lib/internal/streams/readable.js

@@ -1453,16 +1453,9 @@ function createAsyncIterator(stream, options) {
   function finalize() {
     completed = true;
 
-    const preserveHalfOpenDuplex =
-      error === null &&
-      stream.allowHalfOpen === true &&
-      stream.writable === true &&
-      stream.writableEnded !== true;
-
     if (
       (error || options?.destroyOnReturn !== false) &&
-      (error === undefined || stream._readableState.autoDestroy) &&
-      !preserveHalfOpenDuplex
+      (error === undefined || stream._readableState.autoDestroy)
     ) {
       destroyImpl.destroyer(stream, null);
     } else {

test/parallel/test-net-allow-half-open-async-iter-destroy.js

@@ -0,0 +1,53 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const net = require('net');
+
+(async function() {
+  let resolveServerSocket;
+  const serverSocketPromise = new Promise((resolve) => {
+    resolveServerSocket = resolve;
+  });
+
+  const server = net.createServer({
+    allowHalfOpen: true,
+  }, common.mustCall((socket) => {
+    resolveServerSocket(socket);
+  }));
+
+  server.on('error', common.mustNotCall());
+  server.on('close', common.mustCall());
+
+  await new Promise((resolve) => {
+    server.listen(0, common.localhostIPv4, resolve);
+  });
+
+  const clientSocket = net.createConnection({
+    port: server.address().port,
+    host: server.address().address,
+  });
+  clientSocket.on('error', common.mustNotCall());
+
+  await new Promise((resolve) => {
+    clientSocket.end('data', resolve);
+  });
+
+  const serverSocket = await serverSocketPromise;
+
+  let serverRead = '';
+  for await (const chunk of serverSocket) {
+    serverRead += chunk;
+  }
+
+  const destroyed = serverSocket.destroyed;
+  serverSocket.destroy();
+  clientSocket.destroy();
+
+  await new Promise((resolve) => {
+    server.close(resolve);
+  });
+
+  assert.strictEqual(serverRead, 'data');
+  assert.strictEqual(destroyed, true);
+})().then(common.mustCall());

test/parallel/test-net-allow-half-open-async-iter.js

@@ -49,7 +49,7 @@ const net = require('net');
   });
 
   let serverRead = '';
-  for await (const chunk of serverSocket) {
+  for await (const chunk of serverSocket.iterator({ destroyOnReturn: false })) {
     serverRead += chunk;
   }
 

test/parallel/test-stream-readable-async-iter-half-open-duplex.js

@@ -26,7 +26,7 @@ const {
 
   (async () => {
     let read = '';
-    for await (const chunk of duplex) {
+    for await (const chunk of duplex.iterator({ destroyOnReturn: false })) {
       read += chunk;
     }