fs: fix close listener leak in FileHandle streams

lib/internal/fs/streams.js

@@ -152,7 +152,19 @@ function importFd(stream, options) {
     stream[kHandle] = options.fd;
     stream[kFs] = FileHandleOperations(stream[kHandle]);
     stream[kHandle][kRef]();
-    options.fd.on('close', FunctionPrototypeBind(stream.close, stream));
+
+    const onclose = FunctionPrototypeBind(stream.close, stream);
+    options.fd.on('close', onclose);
+    if (options.autoClose === false) {
+      function cleanup() {
+        options.fd.removeListener('close', onclose);
+        options.fd[kUnref]();
+      }
+      stream.once('end', cleanup);
+      stream.once('finish', cleanup);
+      stream.once('error', cleanup);
+    }
+
     return options.fd.fd;
   }
 

test/parallel/test-fs-promises-file-handle-stream.js

@@ -42,7 +42,46 @@ async function validateRead() {
   );
 }
 
+async function validateReusedCreateReadStream() {
+  const filePath = path.resolve(tmpDir, 'tmp-reused-stream.txt');
+  fs.writeFileSync(filePath, Buffer.from('ab', 'utf8'));
+
+  const fileHandle = await open(filePath, 'r');
+  try {
+    await buffer(fileHandle.createReadStream({
+      start: 0,
+      end: 0,
+      autoClose: false,
+    }));
+    assert.strictEqual(fileHandle.listenerCount('close'), 0);
+
+    await buffer(fileHandle.createReadStream({
+      start: 1,
+      end: 1,
+      autoClose: false,
+    }));
+    assert.strictEqual(fileHandle.listenerCount('close'), 0);
+  } finally {
+    await fileHandle.close();
+  }
+}
+
+async function validateReusedCreateWriteStream() {
+  const filePath = path.resolve(tmpDir, 'tmp-reused-write-stream.txt');
+  const fileHandle = await open(filePath, 'w');
+  try {
+    const stream = fileHandle.createWriteStream({ autoClose: false });
+    stream.end('a');
+    await finished(stream);
+    assert.strictEqual(fileHandle.listenerCount('close'), 0);
+  } finally {
+    await fileHandle.close();
+  }
+}
+
 Promise.all([
   validateWrite(),
   validateRead(),
+  validateReusedCreateReadStream(),
+  validateReusedCreateWriteStream(),
 ]).then(common.mustCall());