vfs: fix VirtualReadStream race with async iteration

lib/internal/vfs/streams.js

@@ -80,16 +80,7 @@ class VirtualReadStream extends Readable {
     this.#autoClose = options.autoClose !== false;
 
     if (fd !== null && fd !== undefined) {
-      // Use the already-open file descriptor
       this.#fd = fd;
-      process.nextTick(() => {
-        this.emit('open', this.#fd);
-        this.emit('ready');
-      });
-    } else {
-      // Open the file on next tick so listeners can be attached.
-      // Note: #openFile will not throw - if it fails, the stream is destroyed.
-      process.nextTick(() => this.#openFile());
     }
   }
 
@@ -101,18 +92,16 @@ class VirtualReadStream extends Readable {
     return this.#path;
   }
 
-  /**
-   * Opens the virtual file.
-   * Events are emitted synchronously within this method, which runs
-   * asynchronously via process.nextTick - matching real fs behavior.
-   */
-  #openFile() {
+  _construct(callback) {
     try {
-      this.#fd = this.#vfs.openSync(this.#path);
+      if (this.#fd === null) {
+        this.#fd = this.#vfs.openSync(this.#path);
+      }
+      callback();
       this.emit('open', this.#fd);
       this.emit('ready');
     } catch (err) {
-      this.destroy(err);
+      callback(err);
     }
   }
 

test/parallel/test-vfs-stream-explicit-fd.js

@@ -32,6 +32,18 @@ function readStream(stream) {
   }));
 }
 
+// ReadStream with explicit fd via for-await; autoClose:false leaves fd open
+(async () => {
+  const fd = myVfs.openSync('/file.txt', 'r');
+  const stream = myVfs.createReadStream('/file.txt', { fd, autoClose: false });
+  const chunks = [];
+  for await (const chunk of stream) {
+    chunks.push(chunk);
+  }
+  assert.strictEqual(Buffer.concat(chunks).toString(), 'hello world');
+  myVfs.closeSync(fd);
+})().then(common.mustCall());
+
 // WriteStream with explicit fd; autoClose:false leaves the fd open
 (async () => {
   const fd = myVfs.openSync('/fd-write.txt', 'w');

test/parallel/test-vfs-streams.js

@@ -229,6 +229,18 @@ const { pipeline } = require('stream/promises');
   }));
 }
 
+// createReadStream via for-await iteration
+(async () => {
+  const myVfs = vfs.create();
+  myVfs.writeFileSync('/iter.txt', 'abc');
+  const rs = myVfs.createReadStream('/iter.txt', { encoding: 'utf8' });
+  const chunks = [];
+  for await (const chunk of rs) {
+    chunks.push(chunk);
+  }
+  assert.strictEqual(chunks.join(''), 'abc');
+})().then(common.mustCall());
+
 // start: beyond file size → empty stream
 {
   const myVfs = vfs.create();
@@ -257,6 +269,17 @@ const { pipeline } = require('stream/promises');
   assert.strictEqual(myVfs.readFileSync('/out.txt', 'utf8'), 'hello world');
 })().then(common.mustCall());
 
+// Pipeline read
+(async () => {
+  const myVfs = vfs.create();
+  myVfs.writeFileSync('/in.txt', 'hello world');
+  await pipeline(
+    myVfs.createReadStream('/in.txt'),
+    myVfs.createWriteStream('/copied.txt'),
+  );
+  assert.strictEqual(myVfs.readFileSync('/copied.txt', 'utf8'), 'hello world');
+})().then(common.mustCall());
+
 // Pipeline write with start position
 (async () => {
   const myVfs = vfs.create();